Thomas Mueller's questions

我是 Apache Flink 新手。我尝试使用 Flink 中的 KafkaSource 从 Apache Kafka 获取事件。到目前为止一切顺利，似乎运行良好。重新启动 flink 任务后，我再次收到相同的消息，尽管我已设置 GroupId。

下面是我从 Kafka 读取的代码片段：

KafkaSource<BestellungEvent> bestellungEventSource = KafkaSource.<BestellungEvent>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("bestellungen")
                .setGroupId("bla2")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setDeserializer(KafkaRecordDeserializationSchema.of(new BestellungEventKeyValueDeserializationSchema()))
                .build(); 

有人能帮我解决这个问题吗？

此致

托马斯