Flink KafkaSource从该主题读取所有消息

Question

我的目标是使用Flink KafkaSource阅读来自Kafka主题的所有消息。我尝试用批处理和流模式执行。问题如下:当我将env.setParallelism设置为高于2时，我必须使用包含bug的接收器。于是，我设置了例如:< code > 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是本地环境：StreamExecution环境env=StreamExecutIONEnvironmen.createLocalEnvironmend（）

此时，主题内已经有超过1000万条消息。当作业执行时，我可以在日志中看到:

    [ -> 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

然后他们总共消耗了大约100万条消息并停止了消费，对于所有3条消息，我可以看到：

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

因此，他们并没有完全理解这个话题，只是其中的一部分。当石英作业重新触发时，它再次开始从OffsetsInitializer读取。在earlish（）中，它们使用重复的消息，也使用新消息，不仅是新添加到主题中的消息，还有一些在前一次执行期间未使用的消息。

我还尝试重新命名消费群体，以消除消费者在前一次消费后promise的情况下的补偿问题。

我的问题是 - 我如何配置数据流以完全阅读主题。我的问题与 setParallelism（1） 设置或一般的并行性、使用者组配置或其他任何东西有什么关系？请给我任何关于解决问题的建议。




Answer 1

问题与

builder.setBounded(OffsetsInitializer.latest());

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