Apache Flink Kafka源properties.group.id与Kafka消费者组的疑问

我创建了如下带Kafka源的Flink表：

CREATE TABLE click_event_source (
    cnt_id STRING,
    cnt_type STRING,
    ua STRING,
    en STRING,
    event_time TIMESTAMP_LTZ(6)
) WITH (
    'connector' = 'kafka',
    'topic' = 'bdp.ub',
    'properties.bootstrap.servers' = '<server addresses>',
    'properties.group.id' = 'flink_bdp_test',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset',
    'json.timestamp-format.standard' = 'ISO-8601',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);

我认为properties.group.id应对应Kafka消费者组名称，但执行命令

kafka-consumer-groups.sh --bootstrap-server <server address> --list

列出所有消费者组时，未找到flink_bdp_test组，不过该表确实能从Kafka获取消息。恳请帮忙澄清properties.group.id的含义，谢谢！

1. properties.group.id的真实作用 #

Flink Kafka连接器里的properties.group.id确实是配置项，但它不会直接作为可见的Kafka消费者组出现在集群的组列表中，核心原因是Flink的消费机制特性：

• Flink默认依靠自身的分布式状态后端（内存、RocksDB等）管理消费偏移量，不会将偏移量提交到Kafka的__consumer_offsets主题。Kafka集群只有在接收到偏移量提交请求时，才会记录该消费组的存在，所以用kafka-consumer-groups.sh查不到。
• 这个配置更多是Flink内部标识消费任务的分组，用于逻辑上区分不同的Flink消费作业。

2. 为何能正常消费消息 #

即便Kafka里看不到该消费组，Flink依然可以正常消费：

• 你配置了scan.startup.mode = 'latest-offset'，Flink启动时直接从Kafka主题的最新偏移量位置开始拉取数据，不需要依赖Kafka存储的消费组偏移量。
• 消费过程中的偏移量由Flink自己维护在状态里，和Kafka的消费组管理体系完全独立。

3. 如何让消费组在Kafka中可见 #

如果需要让flink_bdp_test出现在Kafka消费者组列表中，需要让Flink将偏移量提交到Kafka，有两种方式：

• 开启Kafka自动提交偏移量（不推荐生产环境使用，易出现偏移量与Flink状态不一致的问题），在WITH子句中添加：

  'properties.enable.auto.commit' = 'true',
  'properties.auto.commit.interval.ms' = '5000'
  

• 结合Flink检查点机制提交偏移量（推荐），添加配置：

  'kafka.offset.commit.mode' = 'checkpoint'
  

  这种方式下，Flink会在检查点完成时将偏移量提交到Kafka，此时就能通过kafka-consumer-groups.sh看到对应的消费组。

内容的提问来源于stack exchange，提问作者Joe Lin