### Question:
KafkaMessageStream implements TrackingEventStream { @Override public Optional<TrackedEventMessage<?>> peek() { return Optional.ofNullable( peekedEvent == null && !hasNextAvailable(0, TimeUnit.NANOSECONDS) ? null : peekedEvent.value() ); } @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 testPeekOnAnEmptyStreamShouldContainNoElement() { assertThat(emptyStream().peek().isPresent()).isFalse(); }
@Test public void testPeekOnNonEmptyStreamShouldContainSomeElement() throws InterruptedException { GenericTrackedDomainEventMessage<String> firstMessage = trackedDomainEvent("foo"); KafkaMessageStream testSubject = stream(Arrays.asList(firstMessage, trackedDomainEvent("bar"))); assertThat(testSubject.peek().isPresent()).isTrue(); assertThat(testSubject.peek().get()).isEqualTo(firstMessage); }
@Test public void testPeekOnAnInterruptedStreamShouldThrowException() throws InterruptedException { try { KafkaMessageStream testSubject = stream( singletonList(trackedDomainEvent("foo"))); Thread.currentThread().interrupt(); assertThat(testSubject.peek().isPresent()).isFalse(); } finally { Thread.interrupted(); } }