Flink Kafkasource阅读了该主题的所有消息

Question

我的目标是使用Flink Kafkasource阅读Kafka主题的所有消息。我尝试使用批处理和流模式执行。问题是以下： 当我设置env.setParallelism高于2时，我必须使用一个包含错误的接收器。因此，我设置了：例如： streamExecutionEnvironment.setParallelism（1）;

我想要消费的Kafka主题包含3个分区。 这是我拥有的代码片段：

KafkaSourceBuilder<Request> builder = KafkaSource.builder();
    builder.setBootstrapServers(kafkaBrokers);
    builder.setProperty("partition.discovery.interval.ms", "10000");
    builder.setTopics(topic);
    builder.setGroupId(groupId);
    builder.setBounded(OffsetsInitializer.latest());
    builder.setStartingOffsets(OffsetsInitializer.earliest());
    builder.setDeserializer(KafkaRecordDeserializationSchema.of(deserializer));

DataStreamSource<Request> streamSource = streamExecutionEnvironment.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
    streamSource.map(new MyMapper())
            .addSink(new Sink(props)).setParallelism(3) //by this setting I expected to have exactly 3 consumers - per partition/split, but it doesn't work when I do not set anything as well
            .name("Flink " + context.getJobDetail().getKey());

应该在Spring Boot应用程序中运行该代码，该代码将被扩展到该应用程序中，我配置了一个石英作业，该Quartz作业将定期执行，而StreamExecutionEnvironment是本地环境： streamExecutionEnvironment env = StreamExecutionEnvironment.CreateLocalenvironment（）;

此时，该主题中已经有消息，超过10m。 当作业执行时，我可以在日志中看到：

    [ -> Map (1/1)#0] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=rrequest_consumer_test-0, groupId=request_consumer_test] Seeking to EARLIEST offset of partition request-2
INFO 7748 --- [ -> Map (1/1)#0] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=request_consumer_test-0, groupId=request_consumer_test] Seeking to EARLIEST offset of partition request-0
INFO 7748 --- [ -> Map (1/1)#0] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=request_consumer_test-0, groupId=request_consumer_test] Seeking to EARLIEST offset of partition request-1

然后，他们总共消耗了大约1M消息并停止消耗，并且在3个中，我可以看到：

[ -> Map (1/1)#0] o.a.f.c.b.s.reader.fetcher.SplitFetcher  : Finished reading from splits [request-1]

因此，他们没有完全消耗该主题，只有一部分。 重新触发了石英作业时，它再次开始从OffsetSinitializer.earliest（）开始阅读，它们会消耗重复消息，而且还要读取新消息，不仅是新添加到主题中，而且在先前执行过程中没有消耗的消息。

我还尝试过重命名消费者群体，以消除消费者的问题，以防消费者在上一次消费后承诺。

我的问题是 - 如何配置数据流以充分阅读主题。我的问题与setParalalisl（1）设置或并行性有关，消费者组配置或其他任何其他问题？ 请给我任何有关解决问题的建议。

Answer 1

该问题与

builder.setBounded(OffsetsInitializer.latest());

此行有关，告诉Kafka将消息读取工作开始时的最后一个偏移。然后，它将停止消耗更多消息。