@Test @Timeout(value = 2500, unit = TimeUnit.MILLISECONDS) void testStartFetcherWithNullTokenShouldStartFromBeginning() throws InterruptedException { int expectedNumberOfMessages = 1; CountDownLatch messageCounter = new CountDownLatch(expectedNumberOfMessages); SortedKafkaMessageBuffer<KafkaEventMessage> testBuffer = new LatchedSortedKafkaMessageBuffer<>(expectedNumberOfMessages, messageCounter); testSubject.poll( mockConsumer(), new TrackingRecordConverter<>(new ConsumerRecordConverter(), KafkaTrackingToken.emptyToken()), testBuffer::putAll ); messageCounter.await(); assertThat(testBuffer.size()).isEqualTo(expectedNumberOfMessages); }
@Override public Registration poll(Consumer<K, V> consumer, RecordConverter<K, V, E> recordConverter, EventConsumer<E> eventConsumer) { FetchEventsTask<K, V, E> fetcherTask = new FetchEventsTask<>(consumer, pollTimeout, recordConverter, eventConsumer, activeFetchers::remove); activeFetchers.add(fetcherTask); executorService.execute(fetcherTask); return () -> { fetcherTask.close(); return true; }; }
AsyncFetcher implements Fetcher<K, V, E> { @Override public Registration poll(Consumer<K, V> consumer, RecordConverter<K, V, E> recordConverter, EventConsumer<E> eventConsumer) { FetchEventsTask<K, V, E> fetcherTask = new FetchEventsTask<>(consumer, pollTimeout, recordConverter, eventConsumer, activeFetchers::remove); activeFetchers.add(fetcherTask); executorService.execute(fetcherTask); return () -> { fetcherTask.close(); return true; }; } }
AsyncFetcher implements Fetcher<K, V, E> { @Override public Registration poll(Consumer<K, V> consumer, RecordConverter<K, V, E> recordConverter, EventConsumer<E> eventConsumer) { FetchEventsTask<K, V, E> fetcherTask = new FetchEventsTask<>(consumer, pollTimeout, recordConverter, eventConsumer, activeFetchers::remove); activeFetchers.add(fetcherTask); executorService.execute(fetcherTask); return () -> { fetcherTask.close(); return true; }; } @SuppressWarnings("WeakerAccess") protected  AsyncFetcher(Builder<K, V, E> builder); }
AsyncFetcher implements Fetcher<K, V, E> { @Override public Registration poll(Consumer<K, V> consumer, RecordConverter<K, V, E> recordConverter, EventConsumer<E> eventConsumer) { FetchEventsTask<K, V, E> fetcherTask = new FetchEventsTask<>(consumer, pollTimeout, recordConverter, eventConsumer, activeFetchers::remove); activeFetchers.add(fetcherTask); executorService.execute(fetcherTask); return () -> { fetcherTask.close(); return true; }; } @SuppressWarnings("WeakerAccess") protected  AsyncFetcher(Builder<K, V, E> builder); static Builder<K, V, E> builder(); @Override Registration poll(Consumer<K, V> consumer,
                             RecordConverter<K, V, E> recordConverter,
                             EventConsumer<E> eventConsumer); @Override void shutdown(); }
AsyncFetcher implements Fetcher<K, V, E> { @Override public Registration poll(Consumer<K, V> consumer, RecordConverter<K, V, E> recordConverter, EventConsumer<E> eventConsumer) { FetchEventsTask<K, V, E> fetcherTask = new FetchEventsTask<>(consumer, pollTimeout, recordConverter, eventConsumer, activeFetchers::remove); activeFetchers.add(fetcherTask); executorService.execute(fetcherTask); return () -> { fetcherTask.close(); return true; }; } @SuppressWarnings("WeakerAccess") protected  AsyncFetcher(Builder<K, V, E> builder); static Builder<K, V, E> builder(); @Override Registration poll(Consumer<K, V> consumer,
                             RecordConverter<K, V, E> recordConverter,
                             EventConsumer<E> eventConsumer); @Override void shutdown();  }