@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(); }
public void close() { logger.info("Closing down FetchEventsTask using Consumer [{}]", consumer); this.running.set(false); }
FetchEventsTask implements Runnable { public void close() { logger.info("Closing down FetchEventsTask using Consumer [{}]", consumer); this.running.set(false); } }
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); }
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(); }
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();  }