Flink Kafka连接器至eventhub

Question

我正在使用Apache Flink，并尝试通过使用Apache Kafka协议从它接收消息来连接到Azure eventhub。我设法连接到Azure eventhub并接收消息，但我不能使用这里（https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#kafka-consumers-start-position-configuration）所述的flink功能“setStartFromTimestamp（...）”。当我尝试从时间戳获取一些消息时，Kafka说代理端的消息格式是0.10.0之前的。有人面临这个吗？Apache Kafka客户端版本是2.0.1 Apache Flink版本是1.7.2

更新：尝试使用Azure事件中心快速启动示例(https://github.com/Azure/azure-event-hubs-for-kafka/tree/master/quickstart/java)在消费者软件包中，添加了代码以获得带有时间戳的偏移量，如果消息版本低于0.10.0 kafka版本，它将按预期返回null。

        List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC);
        List<TopicPartition> topicPartitions = partitionInfos.stream().map(pi -> new TopicPartition(pi.topic(), pi.partition())).collect(Collectors.toList());
        Map<TopicPartition, Long> topicPartitionToTimestampMap = topicPartitions.stream().collect(Collectors.toMap(tp -> tp, tp -> 0L));
        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestamp = consumer.offsetsForTimes(topicPartitionToTimestampMap);
        System.out.println(offsetAndTimestamp);

Answer 1

很抱歉我们错过了这个。EH现在支持Kafka offsetsForTimes（）（以前不支持）。

以后可以针对我们的Github提出问题。https://github.com/Azure/azure-event-hubs-for-kafka