### Question:
KafkaMessageStream implements TrackingEventStream { @Override public void close() { if (closeHandler != null) { closeHandler.close(); } } @SuppressWarnings("WeakerAccess")  KafkaMessageStream(Buffer<KafkaEventMessage> buffer, Registration closeHandler); @Override Optional<TrackedEventMessage<?>> peek(); @Override boolean hasNextAvailable(int timeout, TimeUnit unit); @Override TrackedEventMessage<?> nextAvailable(); @Override void close();  }### Answer:
@Test public void testClosingMessageStreamShouldInvokeTheCloseHandler() { Registration closeHandler = mock(Registration.class); KafkaMessageStream mock = new KafkaMessageStream(new SortedKafkaMessageBuffer<>(), closeHandler); verify(closeHandler, never()).close(); mock.close(); verify(closeHandler).close(); }