### Question:
FetchEventsTask implements Runnable { public void close() { logger.info("Closing down FetchEventsTask using Consumer [{}]", consumer); this.running.set(false); } FetchEventsTask(Consumer<K, V> consumer,
                    Duration pollTimeout,
                    RecordConverter<K, V, E> recordConverter,
                    EventConsumer<E> eventConsumer,
                    java.util.function.Consumer<FetchEventsTask<K, V, E>> closeHandler); @Override void run(); void close();  }### Answer:
@Test void testFetchEventsTaskInterruptionClosesAsExpected() throws InterruptedException { Thread taskRunner = new Thread(testSubject); taskRunner.start(); doThrow(new InterruptedException()).when(testEventConsumer).consume(any()); assertWithin(Duration.ofMillis(TIMEOUT_MILLIS), () -> assertTrue(expectedToBeClosed.get())); verify(testConsumer).close(); }
@Test void testCloseCallsProvidedCloseHandler() { Thread taskRunner = new Thread(testSubject); taskRunner.start(); testSubject.close(); assertWithin(Duration.ofMillis(TIMEOUT_MILLIS), () -> assertTrue(expectedToBeClosed.get())); verify(testConsumer, timeout(TIMEOUT_MILLIS)).close(); }