@Test void testOpeningMessageStreamWithValidTokenShouldStartTheFetcher() { AtomicBoolean closed = new AtomicBoolean(false); when(fetcher.poll(eq(mockConsumer), any(), any())).thenReturn(() -> { closed.set(true); return true; }); BlockingStream<TrackedEventMessage<?>> result = testSubject.openStream(emptyToken()); verify(consumerFactory).createConsumer(GROUP_ID_PREFIX + GROUP_ID_SUFFIX); verify(fetcher).poll(eq(mockConsumer), any(), any()); result.close(); assertTrue(closed.get()); }
@Override public BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken) { KafkaTrackingToken token = KafkaTrackingToken.from(trackingToken); TrackingRecordConverter<K, V> recordConverter = new TrackingRecordConverter<>(messageConverter, token); String groupId = buildConsumerGroupId(); logger.debug("Consumer Group Id [{}] will start consuming from topics [{}]", groupId, topics); Consumer<K, V> consumer = consumerFactory.createConsumer(groupId); consumer.subscribe( topics, new TrackingTokenConsumerRebalanceListener<>(consumer, recordConverter::currentToken) ); Buffer<KafkaEventMessage> buffer = bufferFactory.get(); Registration closeHandler = fetcher.poll(consumer, recordConverter, buffer::putAll); return new KafkaMessageStream(buffer, closeHandler); }
StreamableKafkaMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken) { KafkaTrackingToken token = KafkaTrackingToken.from(trackingToken); TrackingRecordConverter<K, V> recordConverter = new TrackingRecordConverter<>(messageConverter, token); String groupId = buildConsumerGroupId(); logger.debug("Consumer Group Id [{}] will start consuming from topics [{}]", groupId, topics); Consumer<K, V> consumer = consumerFactory.createConsumer(groupId); consumer.subscribe( topics, new TrackingTokenConsumerRebalanceListener<>(consumer, recordConverter::currentToken) ); Buffer<KafkaEventMessage> buffer = bufferFactory.get(); Registration closeHandler = fetcher.poll(consumer, recordConverter, buffer::putAll); return new KafkaMessageStream(buffer, closeHandler); } }
StreamableKafkaMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken) { KafkaTrackingToken token = KafkaTrackingToken.from(trackingToken); TrackingRecordConverter<K, V> recordConverter = new TrackingRecordConverter<>(messageConverter, token); String groupId = buildConsumerGroupId(); logger.debug("Consumer Group Id [{}] will start consuming from topics [{}]", groupId, topics); Consumer<K, V> consumer = consumerFactory.createConsumer(groupId); consumer.subscribe( topics, new TrackingTokenConsumerRebalanceListener<>(consumer, recordConverter::currentToken) ); Buffer<KafkaEventMessage> buffer = bufferFactory.get(); Registration closeHandler = fetcher.poll(consumer, recordConverter, buffer::putAll); return new KafkaMessageStream(buffer, closeHandler); } protected  StreamableKafkaMessageSource(Builder<K, V> builder); }
StreamableKafkaMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken) { KafkaTrackingToken token = KafkaTrackingToken.from(trackingToken); TrackingRecordConverter<K, V> recordConverter = new TrackingRecordConverter<>(messageConverter, token); String groupId = buildConsumerGroupId(); logger.debug("Consumer Group Id [{}] will start consuming from topics [{}]", groupId, topics); Consumer<K, V> consumer = consumerFactory.createConsumer(groupId); consumer.subscribe( topics, new TrackingTokenConsumerRebalanceListener<>(consumer, recordConverter::currentToken) ); Buffer<KafkaEventMessage> buffer = bufferFactory.get(); Registration closeHandler = fetcher.poll(consumer, recordConverter, buffer::putAll); return new KafkaMessageStream(buffer, closeHandler); } protected  StreamableKafkaMessageSource(Builder<K, V> builder); static Builder<K, V> builder(); @Override BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken); }
StreamableKafkaMessageSource implements StreamableMessageSource<TrackedEventMessage<?>> { @Override public BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken) { KafkaTrackingToken token = KafkaTrackingToken.from(trackingToken); TrackingRecordConverter<K, V> recordConverter = new TrackingRecordConverter<>(messageConverter, token); String groupId = buildConsumerGroupId(); logger.debug("Consumer Group Id [{}] will start consuming from topics [{}]", groupId, topics); Consumer<K, V> consumer = consumerFactory.createConsumer(groupId); consumer.subscribe( topics, new TrackingTokenConsumerRebalanceListener<>(consumer, recordConverter::currentToken) ); Buffer<KafkaEventMessage> buffer = bufferFactory.get(); Registration closeHandler = fetcher.poll(consumer, recordConverter, buffer::putAll); return new KafkaMessageStream(buffer, closeHandler); } protected  StreamableKafkaMessageSource(Builder<K, V> builder); static Builder<K, V> builder(); @Override BlockingStream<TrackedEventMessage<?>> openStream(TrackingToken trackingToken);  }