@Test void testOpeningMessageStreamWithInvalidTypeOfTrackingTokenShouldThrowException() { assertThrows(IllegalArgumentException.class, () -> testSubject.openStream(incompatibleTokenType())); }
@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);  }