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_r291063188
##########
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:
I agree it, I changed the method 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:
[email protected]
With regards,
Apache Git Services
