Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/3915#discussion_r116676798
--- Diff:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
---
@@ -290,10 +296,30 @@ public FlinkKafkaConsumerBase(List<String> topics,
KeyedDeserializationSchema<T>
public FlinkKafkaConsumerBase<T> setStartFromGroupOffsets() {
this.startupMode = StartupMode.GROUP_OFFSETS;
this.specificStartupOffsets = null;
+ this.specificStartupDate = null;
return this;
}
/**
+ * Specifies the consumer to start reading partitions from specific
date. The specified date must before curr timestamp.
+ * This lets the consumer ignore any committed group offsets in
Zookeeper / Kafka brokers.
+ *
+ * The consumer will look up the earliest offset whose timestamp is
greater than or equal to the specific date from the kafka.
+ * If there's no such message, the consumer will use the latest offset
to read data from kafka.
+ *
+ * This method does not effect where partitions are read from when the
consumer is restored
+ * from a checkpoint or savepoint. When the consumer is restored from a
checkpoint or
+ * savepoint, only the offsets in the restored state will be used.
+ *
+ * Note: The api is supported by kafka version >= 0.10 only.
+ *
+ * @return The consumer object, to allow function chaining.
+ */
+ public FlinkKafkaConsumerBase<T> setStartFromSpecificDate(Date date) {
+ throw new RuntimeException("This method supports kafka version
>= 0.10 only.");
--- End diff --
If only 0.10 supports this, shouldn't we add it to the
`FlinkKafkaConsumer010` class only?
---
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.
---
