wangzzu commented on a change in pull request #1253: [CALCITE-3073] Support read from special timestamp in KafkaAdapter URL: https://github.com/apache/calcite/pull/1253#discussion_r291221196
########## File path: kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java ########## @@ -77,14 +90,61 @@ if (tableOptions.getConsumerParams() != null) { consumerConfig.putAll(tableOptions.getConsumerParams()); } - Consumer consumer = new KafkaConsumer<>(consumerConfig); - consumer.subscribe(Collections.singletonList(tableOptions.getTopicName())); - + Consumer consumer; + if (tableOptions.getConsumer() != null) { + consumer = tableOptions.getConsumer(); + } else { + consumer = new KafkaConsumer<>(consumerConfig); + } + if (tableOptions.getTimestamp() != -1) { + // seek to special timestamp + consumer.subscribe(Collections.singletonList(tableOptions.getTopicName()), + new ConsumerRebalanceListener() { + @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { + } + + @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { + // consumer can seek to special offset after the consumer join the consumer group + seekToTimestamp(consumer, tableOptions.getTopicName(), + tableOptions.getTimestamp()); + } + }); + } else { + consumer.subscribe(Collections.singletonList(tableOptions.getTopicName())); + } return new KafkaMessageEnumerator(consumer, tableOptions.getRowConverter(), cancelFlag); } }; } + /** + * Seek to special timestamp. + * + * @param consumer KafkaConsumer; + * @param topicName topic to subscribe; + * @param timestamp the timestamp need to seek; + */ + private void seekToTimestamp(Consumer consumer, String topicName, long timestamp) { + if (hasSeekToTimestamp) { Review comment: The method is not thread-safe, and KafkaConsumer itself is not safe for multi-threaded access. And I think we do not need to consider the case of multiple-threads, because KafkaAdapter does not support multi-threaded. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services