zhztheplayer 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_r291164768
##########
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:
Is the implementation thread-safe? Or we don't have to consider the case of
multiple threads?
----------------------------------------------------------------
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
