### Question:
KafkaMessageStream implements TrackingEventStream { @Override public boolean hasNextAvailable(int timeout, TimeUnit unit) { try { return peekedEvent != null || (peekedEvent = buffer.poll(timeout, unit)) != null; } catch (InterruptedException e) { logger.warn("Consumer thread was interrupted. Returning thread to event processor.", e); Thread.currentThread().interrupt(); return false; } } @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 testHasNextAvailableOnAnEmptyStreamShouldContainNoElement() { assertThat(emptyStream().hasNextAvailable(1, DEFAULT_TIMEOUT_UNIT)).isFalse(); }
@Test public void testHasNextAvailableOnNonEmptyStreamShouldContainSomeElement() throws InterruptedException { KafkaMessageStream testSubject = stream(singletonList(trackedDomainEvent("foo"))); assertThat(testSubject.hasNextAvailable(1, DEFAULT_TIMEOUT_UNIT)).isTrue(); }
@Test public void testHasNextOnAnInterruptedStreamShouldThrowAnException() throws InterruptedException { try { KafkaMessageStream testSubject = stream(singletonList(trackedDomainEvent("foo"))); Thread.currentThread().interrupt(); assertThat(testSubject.hasNextAvailable(1, DEFAULT_TIMEOUT_UNIT)).isFalse(); } finally { Thread.interrupted(); } }