Github user zjureel commented on a diff in the pull request:
https://github.com/apache/flink/pull/3915#discussion_r117206216
--- Diff:
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java
---
@@ -187,31 +191,65 @@ public FlinkKafkaConsumer08(List<String> topics,
KeyedDeserializationSchema<T> d
validateAutoOffsetResetValue(props);
}
+ /**
+ * Search offset from timestamp for each topic in kafka. If no offset
exist, use the latest offset.
+ *
+ * @param partitionTimesMap Kafka topic partition and timestamp
+ * @return Kafka topic partition and the earliest offset after the
timestamp. If no offset exist, use the latest offset in kafka
+ */
+ private Map<KafkaTopicPartition, Long>
convertTimestampToOffset(Map<KafkaTopicPartition, Long> partitionTimesMap) {
--- End diff --
Indeed, user may be doubt about the new method when he used Kafka version
0.8 and 0.9 both. New functionality backwards compatibility is a better
experience, I think this method could be added when timestamp is supported both
by version 0.8 and 0.9
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
