[GitHub] [calcite] danny0405 commented on a change in pull request #1253: [CALCITE-3073] Support read from special timestamp in KafkaAdapter

[GitBox <gi...@apache.org>]

danny0405 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_r291054279
 
 

 ##########
 File path: kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java
 ##########
 @@ -77,14 +87,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
+                  seekOnTimestamp(consumer, tableOptions.getTopicName(),
 
 Review comment:
   Maybe seekToTimestamp is better name.

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services