消息队列Kafka

背景信息

前提条件

注意事项

目前不推荐使用事务写入，这是 Flink 社区和 Kafka 社区的设计缺陷所致。当设置sink.delivery-guarantee = exactly-once，Kafka Connector 会启用事务写入，存在三个已知问题：

• 每个 Checkpoint 会生成一个 Transaction ID。如果 Checkpoint 间隔太短，Transaction ID会过多。Kafka 集群的 Coordinator 可能因此内存不足，从而破坏 Kafka 集群的稳定性。

• 每个事务会创建一个 Producer 实例。如果同时提交的事务太多，TaskManager 的内存可能耗尽，从而破坏 Flink 作业的稳定性。

• 多个 Flink 作业若使用相同的sink.transactional-id-prefix，它们生成的事务 ID 可能冲突。一个作业写入失败时，会阻塞 Kafka 分区的 LSO（Log Start Offset）前进，这会影响所有消费者读取该分区的数据。

如果你需要 Exactly-Once 语义，改用 Upsert Kafka 写入主键表，并用主键保证幂等性。如果需要使用事务写入，请参见EXACTLY_ONCE语义注意事项。

网络连接排查

Flink作业启动时若报错Timed out waiting for a node assignment，通常是因为 Flink 与 Kafka 之间网络不通。

SQL

Kafka连接器可以在SQL作业中使用，作为源表或者结果表。

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

元信息列

CREATE TABLE kafka_source (
  --读取消息所属的topic作为`record_topic`字段
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  --读取ConsumerRecord中的时间戳作为`ts`字段
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  --读取消息的offset作为`record_offset`字段
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  --将`ts`字段中的时间戳作为ProducerRecord的时间戳写入Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

WITH参数

• 通用

  参数	说明	数据类型	是否必填	默认值	备注
  connector	表类型。	String	是	无	固定值为kafka。
  properties.bootstrap.servers	Kafka broker地址。	String	是	无	格式为host:port,host:port,host:port，以英文逗号（,）分割。
  properties.*	对Kafka客户端的直接配置。	String	否	无	后缀名必须是Kafka官方文档中定义的生产者和消费者配置。 Flink会将properties.前缀移除，并将剩余的配置传递给Kafka客户端。例如可以通过'properties.allow.auto.create.topics'='false'来禁用自动创建topic。 不能通过该方式修改以下配置，因为它们会被Kafka连接器覆盖： key.deserializer value.deserializer
  format	读取或写入Kafka消息value部分时使用的格式。	String	否	无	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 说明 更多format参数设置请参见Format参数。
  key.format	读取或写入Kafka消息key部分时使用的格式。	String	否	无	支持的格式 csv json avro debezium-json canal-json maxwell-json avro-confluent raw 说明 使用该配置时，key.options配置是必填的。
  key.fields	Kafka消息key部分对应的源表或结果表字段。	String	否	无	多个字段名以分号（;）分隔。例如field1;field2
  key.fields-prefix	为所有Kafka消息key部分指定自定义前缀，以避免与消息value部分格式字段重名。	String	否	无	该配置项仅用于源表和结果表的列名区分，解析和生成Kafka消息key部分时，该前缀会被移除。 说明 使用该配置时，value.fields-include必须配置为EXCEPT_KEY。
  value.format	读取或写入Kafka消息value部分时使用的格式。	String	否	无	该配置等同于format，只能设置 format 或 value.format 中的一个。如果同时配置，value.format会覆盖format。
  value.fields-include	在解析或生成Kafka消息value部分时，是否要包含消息key部分对应的字段。	String	否	ALL	参数取值如下： ALL（默认值）：所有列都会作为Kafka消息value部分处理 EXCEPT_KEY：除去key.fields定义的字段，剩余字段作为Kafka消息value部分处理

• 源表

  参数	说明	数据类型	是否必填	默认值	备注
  topic	读取的topic名称。	String	否	无	以英文分号 (;) 分隔多个topic名称，例如topic-1和topic-2 说明 topic和topic-pattern两个选项只能指定其中一个。
  topic-pattern	匹配读取topic名称的正则表达式。所有匹配该正则表达式的topic在作业运行时均会被读取。	String	否	无	说明 topic和topic-pattern两个选项只能指定其中一个。
  properties.group.id	消费组ID。	String	否	KafkaSource-{源表表名}	如果指定的group id为首次使用，则必须将properties.auto.offset.reset设置为earliest或latest以指定首次启动位点。
  scan.startup.mode	Kafka读取数据的启动位点。	String	否	group-offsets	取值如下： earliest-offset：从Kafka最早的分区开始读取。 latest-offset：从Kafka最新位点开始读取。 group-offsets（默认值）：从指定的properties.group.id已提交的位点开始读取。 timestamp：从scan.startup.timestamp-millis指定的时间戳开始读取。 specific-offsets：从scan.startup.specific-offsets指定的偏移量开始读取。 说明 该参数在作业无状态启动时生效。作业在从checkpoint重启或状态恢复时，会优先使用状态中保存的进度恢复读取。
  scan.startup.specific-offsets	specific-offsets启动模式下，指定每个分区的启动偏移量。	String	否	无	例如partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	timestamp启动模式下，指定启动位点时间戳。	Long	否	无	单位为毫秒
  scan.topic-partition-discovery.interval	动态检测Kafka topic和partition的时间间隔。	Duration	否	5分钟	分区检查间隔默认为5分钟。需要显式地设置分区检查间隔为非正数才能关闭此功能。开启动态分区发现后，Kafka Source 可以自动地发现新增的分区并自动读取对应分区上的数据。在topic-pattern模式下，不仅读取已有topic的新增分区数据，也会读取符合正则匹配的新增topic的所有分区数据。 说明 在实时计算引擎VVR 6.0.x版本中，动态分区检测默认为关闭。自8.0版本起该功能默认打开，检测间隔默认设置为5分钟。
  scan.header-filter	根据Kafka数据是否包含指定的消息头（Header）对数据进行条件过滤。	String	否	无	Header key和value使用冒号（:）分隔，多个header条件之间使用逻辑运算符（&、|）连接，支持取反逻辑运算符（!）。例如depart:toy|depart:book&!env:test表示保留header中包含depart=toy或depart=book，且不包含env=test的Kafka数据。 说明 仅实时计算引擎VVR 8.0.6及以上版本支持配置该参数。 暂不支持括号运算。 逻辑运算顺序为从左至右。 Header value会以UTF-8格式转换为字符串，与参数指定的header value进行比较。
  scan.check.duplicated.group.id	是否检查通过properties.group.id指定的消费者组有重复。	Boolean	否	false	参数取值如下： true：在启动作业前，系统会检查消费者组是否存在重复。若发现重复，作业将报错并停止运行，从而避免与现有消费者组产生冲突。 false：直接启动作业，不检查消费者组冲突。 说明 仅VVR 6.0.4及以上版本支持该参数。

• 结果表

  参数	说明	数据类型	是否必填	默认值	备注
  topic	写入的topic名称。	String	是	无	无
  sink.partitioner	从Flink并发到Kafka分区的映射模式。	String	否	default	取值如下： default（默认值）：使用Kafka默认的分区模式 fixed：每个Flink并发对应一个固定的Kafka分区。 round-robin：Flink并发中的数据将被轮流分配至Kafka的各个分区。 自定义分区映射模式：如果fixed和round-robin不满足需求，可以创建一个FlinkKafkaPartitioner的子类来自定义分区映射模式。例如org.mycompany.MyPartitioner
  sink.delivery-guarantee	Kafka结果表的语义模式。	String	否	at-least-once	取值如下： none：不保证任何语义，数据可能会丢失或重复。 at-least-once（默认值）：保证数据不丢失，但可能会重复。 exactly-once：使用Kafka事务保证数据不会丢失和重复。 说明 在使用exactly-once语义时，sink.transactional-id-prefix是必填的。
  sink.transactional-id-prefix	在exactly-once语义下使用的Kafka事务ID前缀。	String	否	无	只有sink.delivery-guarantee配置为exactly-once时该配置才会生效。
  sink.parallelism	Kafka结果表算子的并发数。	Integer	否	无	上游算子的并发，由框架决定。

安全与认证

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /*SSL配置*/
  /*配置服务端提供的truststore (CA 证书) 的路径*/
  /*文件管理上传的文件会存放在/flink/usrlib/路径下*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*如果要求客户端认证，则需要配置keystore (私钥) 的路径*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*客户端验证服务器地址的算法，空值表示禁用服务器地址验证*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*SASL配置*/
  /*将SASL机制配置为as SCRAM-SHA-256*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*配置JAAS*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

示例中提到的CA证书和私钥可使用实时计算控制台的文件管理功能上传至平台。上传后文件存放在/flink/usrlib目录下，需要使用的CA证书文件名为my-truststore.jks，则上传后需要在WITH参数中指定'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'来使用该证书。

源表起始位点

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  --从最早位点开始消费
  'scan.startup.mode' = 'earliest-offset',
  --从最末尾位点开始消费
  'scan.startup.mode' = 'latest-offset',
  --从消费者组"my-group"的已提交位点开始消费
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- 如果 "my-group" 为首次使用，则从最早位点开始消费
  'properties.auto.offset.reset' = 'latest', -- 如果 "my-group" 为首次使用，则从最末尾位点开始消费
  --从指定的毫秒时间戳1655395200000开始消费
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  --从指定位点开始消费
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

源表位点提交

Kafka源表只在checkpoint成功后将当前消费位点提交至Kafka集群。如果checkpoint间隔设置较长，在Kafka集群侧观察到的消费位点会有延迟。在进行checkpoint时，Kafka源表会将当前读取进度存储在状态中，并不依赖于提交到集群上的位点进行故障恢复，提交位点仅仅是为了在Kafka侧能够监控到读取进度，位点提交失败不会对数据正确性产生任何影响。

结果表自定义分区器

如果内置的Kafka Producer分区模式无法满足需求，可以实现自定义分区模式将数据写入对应的分区。自定义分区器需要继承FlinkKafkaPartitioner，开发完成后编译JAR包，使用文件管理功能上传至实时计算控制台。上传并引用完成后，请在WITH参数中设置sink.partitioner参数，参数值为分区器完整的类路径，如org.mycompany.MyPartitioner。

Kafka、Upsert Kafka或Kafka JSON catalog的选择

Kafka是一种只能添加数据的消息队列系统，无法进行数据的更新和删除操作，因此在流式SQL计算中无法处理上游的CDC变更数据和聚合、联合等算子的回撤逻辑。如果需要将含有变更或回撤类型的数据写入Kafka，请使用对变更数据进行特殊处理的Upsert Kafka结果表。

为了方便将上游数据库中一个或多个数据表中的变更数据批量同步到Kafka中，可以使用Kafka JSON catalog。如果Kafka中存储的数据格式为JSON，使用Kafka JSON catalog可以省去定义schema和WITH参数的步骤。详情可参见管理Kafka JSON Catalog。

使用示例

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    --可选，将嵌套列全部展开。
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--使用COALESCE处理空值情况。

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "阿里云",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

Datastream API

• 构建Kafka Source

  Kafka Source提供了构建类来创建Kafka Source的实例。我们将通过以下示例代码介绍如何构建Kafka Source来消费input-topic最早位点的数据，消费组名称为my-group，并将Kafka消息体反序列化为字符串。

  Java

  KafkaSource<String> source = KafkaSource.<String>builder()
      .setBootstrapServers(brokers)
      .setTopics("input-topic")
      .setGroupId("my-group")
      .setStartingOffsets(OffsetsInitializer.earliest())
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build();
  
  env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

  在构建KafkaSource时，必须指定以下参数。

  参数	说明
  BootstrapServers	Kafka Broker地址，通过setBootstrapServers(String)方法配置。
  GroupId	消费者组ID，通过setGroupId(String)方法配置。
  Topics或Partition	订阅的Topic或Partition名称。Kafka Source提供了以下三种Topic或Partition的订阅方式： Topic列表，订阅Topic列表中所有Partition。 KafkaSource.builder().setTopics("topic-a","topic-b") 正则表达式匹配，订阅与正则表达式所匹配的Topic下的所有Partition。 KafkaSource.builder().setTopicPattern("topic.*") Partition列表，订阅指定的Partition。 final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList( new TopicPartition("topic-a", 0), // Partition 0 of topic "topic-a" new TopicPartition("topic-b", 5))); // Partition 5 of topic "topic-b" KafkaSource.builder().setPartitions(partitionSet)
  Deserializer	解析Kafka消息的反序列化器。 反序列化器通过setDeserializer(KafkaRecordDeserializationSchema)来指定，其中KafkaRecordDeserializationSchema定义了如何解析Kafka的ConsumerRecord。如果只解析Kafka消息中的消息体（Value）的数据，可以通过以下任何一种方式实现： 使用Flink提供的KafkaSource构建类中的setValueOnlyDeserializer(DeserializationSchema)方法，其中DeserializationSchema定义了如何解析Kafka消息体中的二进制数据。 使用Kafka提供的解析器，包括多种实现类。例如，可以使用StringDeserializer来将Kafka消息体解析成字符串。 import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); 说明 如果需要完整地解析ConsumerRecord，则需要自行实现KafkaRecordDeserializationSchema接口。

  XML

  Maven中央库中已经放置了Kafka DataStream连接器。

  <dependency>
      <groupId>com.alibaba.ververica</groupId>
      <artifactId>ververica-connector-kafka</artifactId>
      <version>${vvr-version}</version>
  </dependency>

  在使用Kafka DataStream连接器时，需要了解以下Kafka属性：

  • 起始消费位点

    Kafka Source能够通过位点初始化器（OffsetsInitializer）来指定从不同的偏移量开始消费。内置的位点初始化器包括以下内容。

    位点初始化器	代码设置
    从最早位点开始消费。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    从最末尾位点开始消费。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    从时间戳大于等于指定时间的数据开始消费，单位为毫秒。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    从消费组提交的位点开始消费，如果提交位点不存在，使用最早位点。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    从消费组提交的位点开始消费，不指定位点重置策略。	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    说明

    • 如果以上内置的初始化器不能满足需求，也可以自己实现自定义的位点初始化器。

    • 如果未指定位点初始化器，则默认使用OffsetsInitializer.earliest()，即最早位点。

  • 流模式和批模式

    Kafka Source支持流式和批式两种运行模式。默认情况下，Kafka Source设置为以流模式运行，因此作业永远不会停止，直到Flink作业失败或被取消。如果要配置Kafka Source在批模式下运行，可以使用setBounded(OffsetsInitializer)指定停止偏移量，当所有分区都达到其停止偏移量时，Kafka Source会退出运行。

    说明

    通常，流模式下Kafka Source没有停止偏移量。为了方便对代码进行调试，流模式下可以使用 setUnbounded(OffsetsInitializer) 指定停止偏移量。请留意指定流模式和批模式停止偏移量的方法名（setUnbounded 和 setBounded）是不同的。

  • 动态分区检查

    为了在不重启Flink作业的情况下，处理Topic扩容或新建Topic等场景，可以在提供的Topic或Partition订阅模式下，启用动态分区检查功能。

    说明

    默认开启动态分区检查功能，分区检查间隔默认为5分钟。需要显式地设置分区检查间隔为非正数才能关闭此功能。代码示例如下。

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // 每10秒检查一次新分区。

    重要

    动态分区检查功能依赖于Kafka集群的元信息更新机制。如果Kafka集群未及时更新分区信息，可能导致新分区无法被发现。请确保Kafka集群的partition.discovery.interval.ms配置与实际情况匹配。

  • 事件时间和水印

    Kafka Source默认使用Kafka消息中的时间戳作为事件时间。可以自定义水印策略（Watermark Strategy）以从消息中提取事件时间，并向下游发送水印。

    env.fromSource(kafkaSource, new CustomWatermarkStrategy(), "Kafka Source With Custom Watermark Strategy")

    如果需要了解自定义水印策略（Watermark Strategy），请参见Generating Watermarks。

    说明

    若并行Source的部分任务长期处于空闲状态（如某个Kafka分区长时间无数据输入，或Source并发数超过Kafka分区数量），可能导致Watermark生成机制失效。此时系统将无法正常触发窗口计算，造成数据处理流程停滞。

    为解决此类问题，可通过以下方式调整：

    • 配置Watermark超时机制：启用table.exec.source.idle-timeout参数，强制系统在指定超时时间后生成Watermark，确保窗口计算周期的推进。

    • 数据源优化：建议保持Kafka分区数与Source并发度的合理比例（推荐分区数≥Source并行度）。

  • 消费位点提交

    Kafka Source在Checkpoint完成时，提交当前的消费位点，以保证Flink的Checkpoint状态和Kafka Broker上的提交位点一致。如果未开启Checkpoint，Kafka Source依赖于Kafka Consumer内部的位点定时自动提交逻辑，自动提交功能由enable.auto.commit和 auto.commit.interval.ms两个Kafka Consumer配置项进行配置。

    说明

    Kafka Source不依赖于Broker上提交的位点来恢复失败的作业。提交位点只是为了上报Kafka Consumer和消费组的消费进度，以在Broker端进行监控。

  • 其他属性

    除了上述属性之外，还可以使用setProperties(Properties) 和setProperty(String, String) 为Kafka Source和Kafka Consumer设置任意属性。KafkaSource通常有以下配置项。

    配置项	说明
    client.id.prefix	指定用于Kafka Consumer的客户端ID前缀。
    partition.discovery.interval.ms	定义Kafka Source检查新分区的时间间隔。 说明 partition.discovery.interval.ms会在批模式下被覆盖为-1。
    register.consumer.metrics	指定是否在Flink中注册Kafka Consumer的指标。
    其他Kafka Consumer配置	Kafka Consumer的配置详情，请参见Apache Kafka。 重要 Kafka Connector会强制覆盖部分手动配置的参数项，覆盖详情如下： key.deserializer始终被覆盖为ByteArrayDeserializer。 value.deserializer始终被覆盖为ByteArrayDeserializer。 auto.offset.reset.strategy被覆盖为OffsetsInitializer#getAutoOffsetResetStrategy()。

    以下示例展示如何配置Kafka Consumer，以使用PLAIN作为SASL机制并提供JAAS配置。

    KafkaSource.builder()
        .setProperty("sasl.mechanism", "PLAIN")
        .setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";")

  • 监控

    Kafka Source在Flink中注册指标，用于监控和诊断。

    • 指标范围

      Kafka source reader的所有指标都注册在KafkaSourceReader指标组下，KafkaSourceReader是operator指标组的子组。与特定主题分区相关的指标注册在KafkaSourceReader.topic.<topic_name>.partition.<partition_id>指标组中。

      例如Topic "my-topic"、分区1的当前消费位点（currentOffset）注册在<some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset。成功提交位点的次数（commitsSucceeded）注册在<some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded。

    • 指标列表

      指标名称	描述	范围
      currentOffset	当前消费位点	TopicPartition
      committedOffset	当前提交位点	TopicPartition
      commitsSucceeded	成功提交的次数	KafkaSourceReader
      commitsFailed	失败的提交次数	KafkaSourceReader

    • Kafka Consumer指标

      Kafka Consumer的指标注册在指标组KafkaSourceReader.KafkaConsumer。例如Kafka Consumer指标records-consumed-total注册在<some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total。

      可以使用配置项register.consumer.metrics配置是否注册Kafka消费者的指标。默认此选项设置为true。对于Kafka Consumer的指标，可参见Apache Kafka。

• 构建Kafka Sink

  Flink Kafka Sink可以实现将流数据写入一个或多个Kafka Topic。

  DataStream<String> stream = ...
  
  
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", );
  KafkaSink<String> kafkaSink =
                  KafkaSink.<String>builder()
                          .setKafkaProducerConfig(kafkaProperties) // // producer config
                          .setRecordSerializer(
                                  KafkaRecordSerializationSchema.builder()
                                          .setTopic("my-topic") // target topic
                                          .setKafkaValueSerializer(StringSerializer.class) // serialization schema
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // fault-tolerance
                          .build();
  
  stream.sinkTo(kafkaSink);

  需要配置以下参数。

  参数	说明
  Topic	数据写入的默认Topic名称。
  数据序列化	构建时需要提供KafkaRecordSerializationSchema来将输入数据转换为 Kafka的ProducerRecord。Flink提供了schema构建器，以提供一些通用的组件，例如消息键（key）/消息体（value）序列化、topic选择、消息分区，同样也可以通过实现对应的接口来进行更丰富的控制。ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp)方法会在每条数据流入的时候调用，来生成ProducerRecord写入Kafka。 用户可以对每条数据如何写入Kafka进行细粒度地控制。通过ProducerRecord可以进行以下操作： 设置写入的Topic名称。 定义消息键（Key）。 指定数据写入的Partition。
  Kafka客户端属性	bootstrap.servers必填，以逗号分隔的Kafka Broker列表。
  容错语义	启用Flink的Checkpoint后，Flink Kafka Sink可以保证精确一次的语义。除了启用Flink的Checkpoint外，还可以通过DeliveryGuarantee参数来指定不同的容错语义，DeliveryGuarantee参数详情如下： DeliveryGuarantee.NONE：（默认设置）Flink不做任何保证，数据可能会丢失或重复。 DeliveryGuarantee.AT_LEAST_ONCE：保证不会丢失任何数据，但可能会重复。 DeliveryGuarantee.EXACTLY_ONCE：使用Kafka事务提供精确一次的语义保证。 说明 使用EXACTLY_ONCE语义时，需要注意的事项请参见EXACTLY_ONCE语义注意事项。

表名与Topic的映射策略

在使用kafka作为数据摄入作业的目标端时，由于写入到Kafka消息格式（debezium-json或者canal-json）中还包含表名信息，后续消费Kafka消息时往往以数据中的表名信息作为实际表名（而非topic名称），因此需要谨慎配置表名与Topic的映射策略。

假设在MySQL有mydb.mytable1,mydb.mytable2两张表需要同步，可能的配置策略有以下几种：

1. 不配置任何映射策略

在没有任何映射策略的情况下，每张表会写入到对应的由“库名.表名”组成的topic中。因此mydb.mytable1的数据会写入到名为mydb.mytable1的topic中，mydb.mytable2的数据会写入到名为mydb.mytable2的topic中。配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

2. 配置route规则进行映射（不推荐）

在很多场景下，用户不希望写入的topic直接为“库名.表名”的格式，希望将数据写入到指定的topic中，因此会配置route规则进行映射。配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}
  
 route:
  - source-table: mydb.mytable1,mydb.mytable2
    sink-table: mytable1

此时所有来自mydb.mytable1,mydb.mytable2的数据都会写入到mytable1这一个topic中。

然而，通过route规则修改写入的topic名称时，也会修改Kafka消息（debezium-json或者canal-json格式）中的表名信息，此时Kafka消息中所有的表名都为mytable1，在其他系统消费这个topic的Kafka消息时，可能出现不符合预期的情况。

3. 配置sink.tableId-to-topic.mapping参数进行映射（推荐）

为了在配置表名与Topic的映射规则的同时保留源表表名信息，可以使用sink.tableId-to-topic.mapping参数完成该需求。配置示例如下：

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

或者

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: mydb.mytable1,mydb.mytable2
  server-id: 8601-8604
  sink.tableId-to-topic.mapping: mydb.mytable1:mytable;mydb.mytable2:mytable

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: ${kafka.bootstraps.server}

此时所有来自mydb.mytable1,mydb.mytable2的数据都会写入到mytable1这一个topic中，并且Kafka消息（debezium-json或者canal-json格式）中的表名信息仍然为mydb.mytable1或者mydb.mytable2，在其他系统消费这个topic的Kafka消息时，能正确获取源表表名信息。

EXACTLY_ONCE语义注意事项

• 配置消费者隔离级别

  所有消费 Kafka 数据的应用必须设置 isolation.level：

  • read_committed：只读取已提交的数据。

  • read_uncommitted（默认）：可读取未提交的数据。

  EXACTLY_ONCE 依赖read_committed。否则消费者可能看到未提交数据，破坏一致性。

• 事务超时与数据丢失

  Flink 从 Checkpoint 恢复时，仅依赖该 Checkpoint 开始前已提交的事务。如果作业崩溃到重启的时间超过 Kafka 事务超时，Kafka 会自动中止事务，导致数据丢失。

  • Kafka Broker 默认transaction.max.timeout.ms = 15 分钟。

  • Flink Kafka Sink 默认设置 transaction.timeout.ms = 1 小时。

  • 你必须在 Broker 端提高transaction.max.timeout.ms，使其不小于 Flink 的设置。

• Producer 池与 Checkpoint 并发

  EXACTLY_ONCE 模式使用固定大小的 Kafka Producer 池。每个 Checkpoint 占用池中的一个 Producer。如果并发 Checkpoint 数超过池大小，作业会失败。

  请根据最大并发 Checkpoint 数调整 Producer 池大小。

• 并行度缩容限制

  如果作业在第一个 Checkpoint 前失败，重启后不会保留原有 Producer 池信息。因此，在第一个 Checkpoint 完成前，不要缩减作业并行度。如必须缩容，并行度不得低于FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR。

• 事务阻塞读取

  在read_committed模式下，任何未结束（未提交也未中止）的事务会阻塞整个 Topic 的读取。

  例如：

  • 事务 1 写入数据。

  • 事务 2 写入并提交数据。

  • 只要事务 1 未结束，事务 2 的数据对消费者不可见。

  因此：

  • 正常运行时，数据可见延迟约等于 Checkpoint 间隔。

  • 作业失败时，正在写入的 Topic 会阻塞消费者，直到作业重启或事务超时，极端情况下，事务超时甚至会影响读取。

常见问题

• Flink如何获取Kafka中的JSON数据？

• 为什么Kafka源表数据基于Event Time的窗口后，不能输出数据？

• 如何连接配置了安全信息的Kafka集群？

• 如何解决字段命名冲突的问题？

• 空间管理与操作