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:
[email protected]
With regards,
Apache Git Services
