Redpanda消费者组与Apache Flink 1.17协同工作异常问题

问题出在哪 #

1. Flink新KafkaSource配置用错了：你用的是Flink 1.13之后的新KafkaSource API，旧API里的commit.offsets.on.checkpoint不能通过setProperty来设置，group.id的用法也和旧API不一样。
2. 多个独立Flink作业没法靠group.id实现负载均衡：每个Flink作业都是独立的分布式应用，就算设了相同的group.id，它们也不会像原生Kafka消费者组那样协作分配分区，反而会各自消费全部分区，结果就是重复收到消息。

怎么修复 #

1. 修正KafkaSource的配置代码 #

新KafkaSource API有专门的方法来设置group.id和提交offset的规则，别再用setProperty了：

KafkaSource<Map<String, Object>> logSource = KafkaSource.<Map<String, Object>>builder()
        .setBootstrapServers(BOOTSTRAP_SERVER)
        .setTopics("logs")
        .setGroupId("group1") // 用新API的专属方法，替代setProperty("group.id", ...)
        .setCommitOffsetsOnCheckpoint(true) // 同样用专属方法，替代旧的配置项
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new LogDeserializer())
        .build();

2. 实现无重复消费的正确姿势 #

如果想要多个消费者分工消费主题、不重复，别启动多个独立的Flink作业，直接调整单个作业的并行度就行：

• 把作业并行度设成和主题分区数相等或者更小
• Flink会自动把主题的分区分配给不同的并行子任务，每个子任务只处理部分分区，自然不会重复消费

3. 非要用多个独立作业的话（不推荐） #

如果因为业务限制必须开多个作业，那就手动给每个作业分配特定的分区，避免重叠：

// 作业1只处理分区0、1
KafkaSource<Map<String, Object>> logSource1 = KafkaSource.<Map<String, Object>>builder()
        .setBootstrapServers(BOOTSTRAP_SERVER)
        .setTopicPartitions(
            Collections.singletonList(new TopicPartition("logs", 0)),
            Collections.singletonList(new TopicPartition("logs", 1))
        )
        .setGroupId("group1")
        .setCommitOffsetsOnCheckpoint(true)
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new LogDeserializer())
        .build();

// 作业2只处理分区2、3
KafkaSource<Map<String, Object>> logSource2 = KafkaSource.<Map<String, Object>>builder()
        .setBootstrapServers(BOOTSTRAP_SERVER)
        .setTopicPartitions(
            Collections.singletonList(new TopicPartition("logs", 2)),
            Collections.singletonList(new TopicPartition("logs", 3))
        )
        .setGroupId("group1")
        .setCommitOffsetsOnCheckpoint(true)
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new LogDeserializer())
        .build();

这种方式需要手动维护分区分配，以后主题加分区了还要同步改作业配置，很麻烦，尽量别用。

额外要检查的点 #

• 确认Redpanda版本和Flink的KafkaSource兼容（Redpanda兼容Kafka协议，一般没问题，但版本差太多可能出问题）
• 检查Flink的checkpoint是不是正常触发：只有checkpoint成功了，offset才会提交到Redpanda里
• 如果用的是Flink 1.12及更早的版本，应该用FlinkKafkaConsumer而不是KafkaSource，那时候group.id和commit.offsets.on.checkpoint的配置是对的，但多个独立作业还是没法实现负载均衡

内容的提问来源于stack exchange，提问作者Krutik Gadhiya