flink如何从kafka读取最新数据

Question

现在，在我的场景中，flink 每次都从 kafka 读取最新数据。 例如， 卡夫卡产品：

日志1

日志2

日志3

读取时，只需要log3。

Kafka 消费者 API，seekToEnd() 可以做到。

FlinkKafkaConsumer 有同样的功能吗？

Answer 1

Flink 1.3 有这个功能。

FlinkKafkaConsumer09 flinkKafkaConsumer09 = new FlinkKafkaConsumer09(properties.getProperty("topic"), new RowDeserializationSchema(properties.getProperty("separator"), resultType), properties);

flinkKafkaConsumer09.setStartFromLatest();