Realtime Compute for Apache Flink:Message Queue for Kafka

Background information

Prerequisites

Precautions

Transactional writes are not recommended because of design limitations in the Flink and Kafka communities. When you set sink.delivery-guarantee = exactly-once, the Kafka connector enables transactional writes, which has the following three known issues:

• Each checkpoint generates a transaction ID. If the checkpoint interval is too short, too many transaction IDs are generated. This can cause the coordinator in the Kafka cluster to run out of memory and compromise the stability of the Kafka cluster.

• Each transaction creates a producer instance. If too many transactions are committed at the same time, the TaskManager may run out of memory and compromise the stability of the Flink job.

• If multiple Flink jobs use the same sink.transactional-id-prefix, the transaction IDs they generate may conflict. When one job fails to write data, it blocks the Log Start Offset (LSO) of the Kafka partition from advancing. This affects all consumers that read data from the partition.

If you require exactly-once semantics, use an Upsert Kafka sink table and use primary keys to ensure idempotence. To use transactional writes, see Precautions for EXACTLY_ONCE semantics.

Troubleshoot network connectivity

If a Flink job reports a Timed out waiting for a node assignment error at startup, it is usually caused by a network connectivity issue between Flink and Kafka.

SQL

You can use the Kafka connector in SQL jobs as a source table or a sink table.

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'
)

Metadata columns

CREATE TABLE kafka_source (
  -- Read the topic of the message as the `record_topic` field.
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` field.
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the offset of the message as the `record_offset` field.
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Write the timestamp from the `ts` field to Kafka as the timestamp of the ProducerRecord.
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

WITH parameters

• Common

  Parameter	Description	Data type	Required	Default value	Remarks
  connector	The table type.	String	Yes	None	The value must be kafka.
  properties.bootstrap.servers	The Kafka broker address.	String	Yes	None	The format is host:port,host:port,host:port. Use commas (,) to separate multiple addresses.
  properties.*	Direct configurations for the Kafka client.	String	No	None	The suffix must be a producer or consumer configuration defined in the official Kafka documentation. Flink removes the properties. prefix and passes the remaining configuration to the Kafka client. For example, you can use 'properties.allow.auto.create.topics'='false' to disable automatic topic creation. Do not use this method to modify the following configurations, because they are overwritten by the Kafka connector: key.deserializer value.deserializer
  format	The format used to read or write the value part of a Kafka message.	String	No	None	Supported formats: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Note For more information about format parameters, see Format Parameters.
  key.format	The format used to read or write the key part of a Kafka message.	String	No	None	Supported formats: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Note If you use this configuration, the key.options configuration is required.
  key.fields	The fields in the source or sink table that correspond to the key part of the Kafka message.	String	No	None	Use semicolons (;) to separate multiple field names. For example: field1;field2
  key.fields-prefix	Specifies a custom prefix for all Kafka message key fields to avoid name conflicts with the value part format fields.	String	No	None	This configuration item is only used to distinguish column names in source and sink tables. The prefix is removed when parsing and generating the key part of the Kafka message. Note If you use this configuration, you must set value.fields-include to EXCEPT_KEY.
  value.format	The format used to read or write the value part of a Kafka message.	String	No	None	This configuration is equivalent to format. You can only set one of format or value.format. If both are configured, value.format overwrites format.
  value.fields-include	Specifies whether to include the fields corresponding to the key part of the message when parsing or generating the value part of the Kafka message.	String	No	ALL	Valid values: ALL (default): All columns are processed as the value part of the Kafka message. EXCEPT_KEY: The remaining fields, excluding those defined in key.fields, are processed as the value part of the Kafka message.

• Source table

  Parameter	Description	Data type	Required	Default value	Remarks
  topic	The name of the topic to read from.	String	No	None	Use semicolons (;) to separate multiple topic names, such as topic-1 and topic-2. Note You can specify only one of the topic and topic-pattern options.
  topic-pattern	A regular expression that matches the names of topics to read from. All topics that match this regular expression are read when the job runs.	String	No	None	Note You can specify only one of the topic and topic-pattern options.
  properties.group.id	The consumer group ID.	String	No	KafkaSource-{source_table_name}	If the specified group ID is used for the first time, you must set properties.auto.offset.reset to earliest or latest to specify the initial start offset.
  scan.startup.mode	The start offset for reading data from Kafka.	String	No	group-offsets	Valid values: earliest-offset: Starts reading from the earliest partition in Kafka. latest-offset: Starts reading from the latest offset in Kafka. group-offsets (default): Starts reading from the committed offset of the specified properties.group.id. timestamp: Starts reading from the timestamp specified by scan.startup.timestamp-millis. specific-offsets: Starts reading from the offset specified by scan.startup.specific-offsets. Note This parameter takes effect when the job starts without a state. When a job restarts from a checkpoint or recovers from a state, it preferentially uses the progress saved in the state to resume reading.
  scan.startup.specific-offsets	In specific-offsets startup mode, specifies the start offset for each partition.	String	No	None	For example: partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	In timestamp startup mode, specifies the start offset timestamp.	Long	No	None	The unit is milliseconds.
  scan.topic-partition-discovery.interval	The interval for dynamically detecting Kafka topics and partitions.	Duration	No	5 minutes	The default partition check interval is 5 minutes. To disable this feature, you must explicitly set the partition check interval to a non-positive value. When dynamic partition discovery is enabled, the Kafka source can automatically discover new partitions and read data from them. In topic-pattern mode, it not only reads data from new partitions of existing topics but also reads data from all partitions of new topics that match the regular expression. Note In VVR 6.0.x, dynamic partition discovery is disabled by default. Starting from VVR 8.0, this feature is enabled by default, and the detection interval is set to 5 minutes.
  scan.header-filter	Conditionally filters data based on whether the Kafka data contains a specified header.	String	No	None	Use a colon (:) to separate the header key and value. Use logical operators (&, |) to connect multiple header conditions. The negation logical operator (!) is also supported. For example, depart:toy|depart:book&!env:test retains Kafka data where the header contains depart=toy or depart=book, and does not contain env=test. Note This parameter is supported only in VVR 8.0.6 and later. Parenthetical operations are not supported. Logical operations are performed from left to right. The header value is converted to a string in UTF-8 format for comparison with the specified header value.
  scan.check.duplicated.group.id	Specifies whether to check for duplicate consumer groups specified by properties.group.id.	Boolean	No	false	Valid values: true: Before starting the job, the system checks for duplicate consumer groups. If a duplicate is found, the job reports an error and stops, avoiding conflicts with existing consumer groups. false: The job starts directly without checking for consumer group conflicts. Note This parameter is supported only in VVR 6.0.4 and later.

• Sink table

  Parameter	Description	Data type	Required	Default value	Remarks
  topic	The name of the topic to write to.	String	Yes	None	None
  sink.partitioner	The mapping mode from Flink concurrency to Kafka partitions.	String	No	default	Valid values: default: Uses the default Kafka partition mode. fixed: Each Flink concurrency corresponds to a fixed Kafka partition. round-robin: Data from Flink concurrencies is allocated to Kafka partitions in a round-robin fashion. Custom partition mapping mode: If fixed and round-robin do not meet your needs, you can create a subclass of FlinkKafkaPartitioner to define a custom partition mapping mode. For example: org.mycompany.MyPartitioner
  sink.delivery-guarantee	The semantic mode of the Kafka sink table.	String	No	at-least-once	Valid values: none: No guarantee is provided. Data may be lost or duplicated. at-least-once (default): Guarantees that no data is lost, but data may be duplicated. exactly-once: Uses Kafka transactions to guarantee that data is not lost or duplicated. Note When using exactly-once semantics, the sink.transactional-id-prefix parameter is required.
  sink.transactional-id-prefix	The prefix for the Kafka transaction ID used in exactly-once semantics.	String	No	None	This configuration takes effect only when sink.delivery-guarantee is set to exactly-once.
  sink.parallelism	The degree of parallelism for the Kafka sink table operator.	Integer	No	None	The degree of parallelism of the upstream operator is determined by the framework.

Security and authentication

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 configuration*/
  /*Configure the path of the truststore (CA certificate) provided by the server.*/
  /*Files uploaded through File Management are stored in the /flink/usrlib/ path.*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*If client authentication is required, configure the path of the keystore (private key).*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*The algorithm for client to verify the server address. An empty value disables server address verification.*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*SASL configuration*/
  /*Configure the SASL mechanism as SCRAM-SHA-256.*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*Configure JAAS*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

You can upload the CA certificate and private key mentioned in the example to the platform using the File Management feature in the Realtime Compute console. After you upload the files, they are stored in the /flink/usrlib directory. If the CA certificate file to be used is named my-truststore.jks, you can specify 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' in the WITH clause to use this certificate.

Source table start offset

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume from the committed offset of the consumer group "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- If "my-group" is used for the first time, consume from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consume from the latest offset.
  -- Consume from the specified millisecond timestamp 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from the specified offset.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Source table offset commit

The Kafka source table commits the current consumer offset to the Kafka cluster only after a checkpoint is successful. If the checkpoint interval is long, the observed consumer offset in the Kafka cluster will be delayed. During a checkpoint, the Kafka source table stores the current read progress in its state and does not rely on the offset committed to the cluster for fault recovery. Committing the offset is only for monitoring the read progress on the Kafka side. A failed offset commit does not affect the correctness of the data.

Sink table custom partitioner

If the built-in Kafka producer partition modes do not meet your needs, you can implement a custom partition mode to write data to the corresponding partitions. A custom partitioner must inherit FlinkKafkaPartitioner. After development, you can compile the JAR package and upload it to the Realtime Compute console using the File Management feature. After you upload and reference the package, you can set the sink.partitioner parameter in the WITH clause to the full class path of the partitioner, such as org.mycompany.MyPartitioner.

Choose between Kafka, Upsert Kafka, or Kafka JSON catalog

Kafka is an append-only message queue system. It does not support data updates or deletions. Therefore, it cannot handle upstream CDC change data or retraction logic from operators such as aggregation and join in streaming SQL. To write data with changes or retractions to Kafka, use an Upsert Kafka sink table, which is designed to handle change data.

To easily synchronize change data from one or more tables in an upstream database to Kafka in batches, you can use a Kafka JSON catalog. If the data stored in Kafka is in JSON format, using a Kafka JSON catalog eliminates the need to define a schema and WITH parameters. For more information, see Manage a Kafka JSON Catalog.

Usage examples

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'
    -- Optional. Flattens all nested columns.
);

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');
-- Use COALESCE to handle null values.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "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

• Build a Kafka source

  The Kafka source provides a builder class to create an instance of the Kafka source. The following example code shows how to build a Kafka source to consume data from the earliest offset of input-topic. The consumer group name is my-group, and the Kafka message body is deserialized as a string.

  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");

  When you build a KafkaSource, you must specify the following parameters.

  Parameter	Description
  BootstrapServers	The Kafka broker address. Configure it using the setBootstrapServers(String) method.
  GroupId	The consumer group ID. Configure it using the setGroupId(String) method.
  Topics or Partition	The name of the subscribed topic or partition. The Kafka source provides the following three ways to subscribe to topics or partitions: Topic list: Subscribes to all partitions in the topic list. KafkaSource.builder().setTopics("topic-a","topic-b") Regular expression matching: Subscribes to all partitions under the topics that match the regular expression. KafkaSource.builder().setTopicPattern("topic.*") Partition list: Subscribes to the specified partitions. 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	The deserializer for parsing Kafka messages. Specify the deserializer using setDeserializer(KafkaRecordDeserializationSchema), where KafkaRecordDeserializationSchema defines how to parse Kafka's ConsumerRecord. If you only need to parse the data in the message body (value) of a Kafka message, you can do so in one of the following ways: Use the setValueOnlyDeserializer(DeserializationSchema) method in the KafkaSource builder class provided by Flink. DeserializationSchema defines how to parse the binary data in the Kafka message body. Use a parser provided by Kafka, which includes multiple implementation classes. For example, you can use StringDeserializer to parse the Kafka message body into a string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Note To fully parse the ConsumerRecord, you must implement the KafkaRecordDeserializationSchema interface yourself.

  XML

  The Kafka DataStream connector is available in the Maven Central Repository.

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

  When you use the Kafka DataStream connector, you must understand the following Kafka properties:

  • Start offset

    The Kafka source can specify to start consuming from different offsets using an offset initializer (OffsetsInitializer). The built-in offset initializers include the following.

    Offset initializer	Code setting
    Consume from the earliest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Consume from the latest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Start consuming from data whose timestamp is greater than or equal to the specified time, in milliseconds.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Start consuming from the offset committed by the consumer group. If the committed offset does not exist, use the earliest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Start consuming from the offset committed by the consumer group, without specifying an offset reset policy.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Note

    • If the built-in initializers do not meet your needs, you can also implement a custom offset initializer.

    • If no offset initializer is specified, OffsetsInitializer.earliest() is used by default, which starts consumption from the earliest offset.

  • Streaming and batch modes

    The Kafka source supports both streaming and batch execution modes. By default, the Kafka source is set to run in streaming mode, so the job never stops until the Flink job fails or is canceled. To configure the Kafka source to run in batch mode, you can use setBounded(OffsetsInitializer) to specify a stop offset. When all partitions reach their stop offset, the Kafka source exits.

    Note

    Typically, the Kafka source has no stop offset in streaming mode. For debugging purposes, you can use setUnbounded(OffsetsInitializer) to specify a stop offset in streaming mode. Note that the method names for specifying the stop offset in streaming and batch modes (setUnbounded and setBounded) are different.

  • Dynamic Partition Check

    To handle scenarios such as topic scale-out or new topic creation without restarting the Flink job, you can enable the dynamic partition discovery feature in the provided topic or partition subscription mode.

    Note

    Dynamic partition discovery is enabled by default, and the partition check interval is 5 minutes. To disable this feature, you must explicitly set the partition check interval to a non-positive value. The following code provides an example.

    KafkaSource.builder()
        .setProperty("partition.discovery.interval.ms", "10000") // Check for new partitions every 10 seconds.

    Important

    The dynamic partition discovery feature relies on the metadata update mechanism of the Kafka cluster. If the Kafka cluster does not update partition information in time, new partitions may not be discovered. Ensure that the partition.discovery.interval.ms configuration of the Kafka cluster matches the actual situation.

  • Event time and watermarks

    By default, the Kafka source uses the timestamp in the Kafka message as the event time. You can customize a watermark strategy to extract the event time from the message and send watermarks downstream.

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

    For more information about custom watermark strategies, see Generating Watermarks.

    Note

    If some tasks of a parallel source are idle for a long time (for example, if a Kafka partition has no data input for a long time, or the degree of parallelism of the source exceeds the number of Kafka partitions), the watermark generation mechanism may fail. In this case, the system cannot trigger window calculations normally, which causes the data processing flow to stall.

    To resolve this issue, you can make the following adjustments:

    • Configure a watermark timeout mechanism: Enable the table.exec.source.idle-timeout parameter to force the system to generate a watermark after a specified timeout period, which ensures the progress of the window calculation cycle.

    • Optimize the data source: Maintain a reasonable ratio between the number of Kafka partitions and the degree of parallelism of the source (recommended: number of partitions ≥ source degree of parallelism).

  • Offset commit

    The Kafka source commits the current consumer offset when a checkpoint is completed to ensure consistency between the Flink checkpoint state and the committed offset on the Kafka broker. If checkpointing is not enabled, the Kafka source relies on the internal automatic offset commit logic of the Kafka consumer. The automatic commit feature is configured by the enable.auto.commit and auto.commit.interval.ms Kafka consumer configuration items.

    Note

    The Kafka source does not rely on the offset committed on the broker to recover from a failed job. Committing the offset is only for reporting the consumption progress of the Kafka consumer and consumer group for monitoring on the broker side.

  • Other properties

    In addition to the preceding properties, you can also use setProperties(Properties) and setProperty(String, String) to set any properties for the Kafka source and Kafka consumer. The KafkaSource typically has the following configuration items.

    Configuration item	Description
    client.id.prefix	Specifies the client ID prefix for the Kafka consumer.
    partition.discovery.interval.ms	Defines the interval at which the Kafka source checks for new partitions. Note partition.discovery.interval.ms is overwritten to -1 in batch mode.
    register.consumer.metrics	Specifies whether to register the Kafka consumer's metrics in Flink.
    Other Kafka consumer configurations	For more information about Kafka consumer configurations, see Apache Kafka. Important The Kafka connector forcibly overwrites some manually configured parameters. The details are as follows: key.deserializer is always overwritten to ByteArrayDeserializer. value.deserializer is always overwritten to ByteArrayDeserializer. auto.offset.reset.strategy is overwritten to OffsetsInitializer#getAutoOffsetResetStrategy().

    The following example shows how to configure a Kafka consumer to use PLAIN as the SASL mechanism and provide a JAAS configuration.

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

  • Monitoring

    The Kafka source registers metrics in Flink for monitoring and diagnosis.

    • Metric scope

      All metrics of the Kafka source reader are registered under the KafkaSourceReader metric group, which is a subgroup of the operator metric group. Metrics related to a specific topic partition are registered in the KafkaSourceReader.topic.<topic_name>.partition.<partition_id> metric group.

      For example, the current consumer offset (currentOffset) for partition 1 of topic "my-topic" is registered under <some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. The number of successful offset commits (commitsSucceeded) is registered under <some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded.

    • Metric list

      Metric name	Description	Scope
      currentOffset	The current consumer offset.	TopicPartition
      committedOffset	The current committed offset.	TopicPartition
      commitsSucceeded	The number of successful commits.	KafkaSourceReader
      commitsFailed	The number of failed commits.	KafkaSourceReader

    • Kafka consumer metrics

      The Kafka consumer's metrics are registered in the KafkaSourceReader.KafkaConsumer metric group. For example, the Kafka consumer metric records-consumed-total is registered under <some_parent_groups>.operator.KafkaSourceReader.KafkaConsumer.records-consumed-total.

      You can use the register.consumer.metrics configuration item to specify whether to register the Kafka consumer's metrics. By default, this option is set to true. For more information about Kafka consumer metrics, see Apache Kafka.

• Build a Kafka sink

  The Flink Kafka sink can write stream data to one or more Kafka topics.

  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);

  You need to configure the following parameters.

  Parameter	Description
  Topic	The default topic name to write data to.
  Data serialization	When building, you need to provide a KafkaRecordSerializationSchema to convert the input data into Kafka's ProducerRecord. Flink provides a schema builder with common components such as message key/value serialization, topic selection, and message partitioning. You can also implement the corresponding interfaces for more control. The ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) method is called for each data record to generate a ProducerRecord to be written to Kafka. You can have fine-grained control over how each data record is written to Kafka. With ProducerRecord, you can perform the following operations: Set the name of the topic to write to. Define the message key. Specify the partition to write data to.
  Kafka client properties	bootstrap.servers is required. It is a comma-separated list of Kafka brokers.
  Fault tolerance semantics	After enabling Flink's checkpointing, the Flink Kafka sink can guarantee exactly-once semantics. In addition to enabling Flink's checkpointing, you can also specify different fault tolerance semantics using the DeliveryGuarantee parameter. The details of the DeliveryGuarantee parameter are as follows: DeliveryGuarantee.NONE: (default) Flink provides no guarantee. Data may be lost or duplicated. DeliveryGuarantee.AT_LEAST_ONCE: Guarantees that no data is lost, but data may be duplicated. DeliveryGuarantee.EXACTLY_ONCE: Uses Kafka transactions to provide exactly-once semantics. Note For precautions when using EXACTLY_ONCE semantics, see Precautions for EXACTLY_ONCE semantics.

Table name and topic mapping strategy

When you use Kafka as the target for a data ingestion job, you must configure the table name and topic mapping strategy carefully. This is because the Kafka message format (debezium-json or canal-json) also contains table name information, and downstream systems often use the table name in the data as the actual table name, not the topic name.

Suppose you need to synchronize two tables, mydb.mytable1 and mydb.mytable2, from MySQL. The possible configuration strategies are as follows:

1. Do not configure any mapping strategy

Without any mapping strategy, each table is written to a topic named "database_name.table_name". Therefore, data from mydb.mytable1 is written to a topic named mydb.mytable1, and data from mydb.mytable2 is written to a topic named mydb.mytable2. The following code provides a configuration example:

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. Configure route rules for mapping (not recommended)

In many scenarios, users do not want the topic to be in the "database_name.table_name" format. They want to write data to a specified topic, so they configure route rules for mapping. The following code provides a configuration example:

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

In this case, all data from mydb.mytable1 and mydb.mytable2 is written to the mytable1 topic.

However, when you use a route rule to change the topic name, it also changes the table name information in the Kafka message (debezium-json or canal-json format). In this case, all table names in the Kafka messages become mytable1, which may cause unexpected behavior when other systems consume Kafka messages from this topic.

3. Configure the sink.tableId-to-topic.mapping parameter for mapping (recommended)

To configure the mapping rule between table names and topics while you retain the source table name information, use the sink.tableId-to-topic.mapping parameter. The following code provides a configuration example:

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}

or

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}

In this case, all data from mydb.mytable1 and mydb.mytable2 is written to the mytable1 topic. The table name information in the Kafka message (debezium-json or canal-json format) remains mydb.mytable1 or mydb.mytable2. When other systems consume Kafka messages from this topic, they can correctly retrieve the source table name information.

Precautions for EXACTLY_ONCE semantics

• Configure consumer isolation level

  All applications that consume Kafka data must set isolation.level:

  • read_committed: Reads only committed data.

  • read_uncommitted (default): Can read uncommitted data.

  EXACTLY_ONCE depends on read_committed. Otherwise, consumers may see uncommitted data, which compromises consistency.

• Transaction timeout and data loss

  When Flink recovers from a checkpoint, it relies only on transactions that were committed before that checkpoint began. If the time from job crash to restart exceeds the Kafka transaction timeout, Kafka automatically aborts the transaction, which causes data loss.

  • The default transaction.max.timeout.ms on the Kafka broker is 15 minutes.

  • The default transaction.timeout.ms for the Flink Kafka sink is 1 hour.

  • You must increase transaction.max.timeout.ms on the broker side to be greater than or equal to the Flink setting.

• Producer pool and checkpoint concurrency

  EXACTLY_ONCE mode uses a fixed-size pool of Kafka producers. Each checkpoint occupies one producer from the pool. If the number of concurrent checkpoints exceeds the pool size, the job will fail.

  Adjust the producer pool size based on the maximum number of concurrent checkpoints.

• Parallelism scale-in limitation

  If the job fails before the first checkpoint, the original producer pool information is not retained upon restart. Therefore, do not reduce the job's degree of parallelism before the first checkpoint is completed. If you must scale in, the degree of parallelism must be greater than or equal to FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transaction blocking reads

  In read_committed mode, any unfinished transaction (neither committed nor aborted) blocks reading from the entire topic.

  For example:

  • Transaction 1 writes data.

  • Transaction 2 writes and commits data.

  • As long as transaction 1 is not finished, the data from transaction 2 is not visible to consumers.

  Therefore:

  • During normal operation, the data visibility latency is approximately equal to the checkpoint interval.

  • When a job fails, the topic being written to will block consumers until the job restarts or the transaction times out. In extreme cases, a transaction timeout can even affect reading.

FAQ

• How does Flink retrieve JSON data from Kafka?

• Why does a Kafka source table with event time-based windows not output data?

• How do I connect to a Kafka cluster with security information configured?

• How do I resolve field naming conflicts?

• Storage management and operations