Realtime Compute for Apache Flink:Message Queue for Kafka

Background information

Prerequisites

Usage notes

Currently, transactional writes are not recommended due to design limitations in the Flink and Kafka communities. When you set sink.delivery-guarantee = exactly-once, the Kafka connector enables transactional writes, which have three known issues:

• Each checkpoint generates a new transaction ID. If the checkpoint interval is too short, the number of transaction IDs becomes excessive. As a result, the Kafka coordinator may run out of memory, compromising Kafka cluster stability.

• Each transaction creates a producer instance. If too many transactions are committed simultaneously, the TaskManager may run out of memory, destabilizing the Flink job.

• If multiple Flink jobs use the same sink.transactional-id-prefix, their generated transaction IDs may conflict. If a job fails to write, it blocks the Log Start Offset (LSO) of the Kafka partition from advancing, affecting all consumers reading from that partition.

If you require exactly-once semantics, use Upsert Kafka to write to a primary key table and rely on the primary key to ensure idempotence. To use transactional writes, see Notes on Exactly-Once Semantics.

Troubleshoot network connectivity

If a Flink job reports the error Timed out waiting for a node assignment during startup, this typically indicates 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 to which the message belongs as the value of the record_topic field.
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp in ConsumerRecord as the value of the ts field.
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the offset of the message as the value of the record_offset field.
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

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

WITH parameters

• General

  parameter	Description	Data type	Required	Default value	Remarks
  connector	The type of the table.	String	Yes	None	Set the value to kafka.
  properties.bootstrap.servers	The IP address and port number of a Kafka broker.	String	Yes	None	Format: host:port,host:port,host:port. Separate multiple host:port pairs with commas (,).
  properties.*	The options that are configured for the Kafka client.	String	No	None	The suffix must comply with the rules that are defined in Producer Configs and Consumer Configs. Flink removes the "properties." prefix and passes the remaining configuration to the Kafka client. For example, you can disable automatic topic creation using 'properties.allow.auto.create.topics'='false'. The following configurations cannot be modified in this way because the Kafka connector overwrites them: key.deserializer value.deserializer
  format	The format used to read or write the value field of Kafka messages.	String	No	No default value.	Supported formats csv json avro debezium-json canal-json maxwell-json avro-confluent raw Note For more information about format options, see Format Options.
  key.format	The format used to read or write the key field of Kafka messages.	String	No	No default value.	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 key fields in the source table or sink table that correspond to the key fields of Kafka messages.	String	No	None	Separate multiple field names with semicolons (;), such as field1;field2.
  key.fields-prefix	The custom prefix for all key fields in Kafka messages. You can configure this option to prevent name conflicts with the value fields.	String	No	None	This option is used only to distinguish the column names of source tables and sink tables. The prefix is removed from the column names when the key fields of Kafka messages are parsed or generated. Note When you use this configuration, you must set the value.fields-include option to EXCEPT_KEY.
  value.format	The format used to read or write the value field of Kafka messages.	String	No	None	This configuration is equivalent to format, and you can configure only one of format or value.format. If you configure both, value.format overwrites format.
  value.fields-include	Specifies whether to include corresponding message keys when parsing or generating Kafka message values.	String	No	ALL	Valid values: ALL (default): All fields are processed as the value of a Kafka message. EXCEPT_KEY: All fields except for the fields specified by the key.fields option are processed as the Kafka message value.

• Source table

  Parameter	Description	Data type	Required	Default value	Remarks
  topic	The name of the topic from which you want to read data.	String	No	None	Separate multiple topic names with semicolons (;), such as topic-1 and topic-2. Note You cannot use the topic option together with the topic-pattern option.
  topic-pattern	The regular expression that is used to match topics. Data of all topics whose names match the specified regular expression is read when a deployment is running.	String	No	None	Note You cannot use the topic option together with the topic-pattern option.
  properties.group.id	The consumer group ID.	String	No	KafkaSource-{Name of the source table}	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 Kafka to read data.	String	No	group-offsets	Valid values: earliest-offset: Kafka reads data from the earliest partition. latest-offset: Kafka reads data from the latest offset. group-offsets (default): reads data from the offset that is committed by the consumer group with the ID that is specified by the properties.group.id option. timestamp: Reads data from the timestamp specified by scan.startup.timestamp-millis. specific-offsets: reads data from the offset that is specified by the scan.startup.specific-offsets option. Note This option takes effect when the deployment is started without states. When the deployment is restarted from a checkpoint or resumes from the specified state, the deployment preferentially starts to read data at the progress that is saved in the state data.
  scan.startup.specific-offsets	The start offset of each partition when the scan.startup.mode option is set to specific-offsets.	String	No	None	Example: partition:0,offset:42;partition:1,offset:300.
  scan.startup.timestamp-millis	The timestamp of the start offset when the scan.startup.mode option is set to timestamp.	Long	No	None	Unit: milliseconds.
  scan.topic-partition-discovery.interval	The time interval for dynamically detecting Kafka topics and partitions.	Duration	No	5 minutes	The default partition discovery interval is 5 minutes. To disable this feature, you must explicitly set this option to a non-positive value. After the dynamic partition discovery feature is enabled, the Kafka source can automatically discover new partitions and read data from the partitions. In topic-pattern mode, the Kafka source reads data from new partitions of existing topics and data from all partitions of new topics that match the regular expression. Note In Realtime Compute for Apache Flink that uses VVR 6.0.X, the dynamic partition discovery feature is disabled by default. In Realtime Compute for Apache Flink that uses VVR 8.0 or later, this feature is enabled by default. The default partition discovery interval is 5 minutes.
  scan.header-filter	Kafka data filtering based on whether the data contains a specific message header.	String	No	None	Separate a header key and the value with a colon (:). Separate multiple headers with logical operators such as AND (&) or OR (|). The logical operator NOT (!) is supported. For example, depart:toy|depart:book&!env:test indicates that the Kafka data whose header contains depart=toy or depart=book and does not contain env=test is retained. Note This option is supported only for Realtime Compute for Apache Flink that uses VVR 8.0.6 or later. Parenthesis operations are not supported. Logical operations are performed from left to right in sequence. The header value in the UTF-8 format is converted into a string and compared with the header value specified by the scan.header-filter option.
  scan.check.duplicated.group.id	Specifies whether to check for duplicate consumer groups specified by the properties.group.id parameter.	Boolean	No	false	Valid values: true: Checks duplicate consumer groups before a job starts. If duplicate consumer groups exist, reports an error and suspends the job to prevent conflicts. false: Does not check duplicate consumer groups before a job starts. Note This option is supported only for Realtime Compute for Apache Flink that uses VVR 6.0.4 or later.

• Sink-specific

  Parameters	Description	Data type	Required	Default value	Remarks
  topic	The name of the topic to which data is written.	String	Yes	None	N/A
  sink.partitioner	The pattern for mapping Flink concurrency to Kafka partitions.	String	No	default	Valid values: default (default): Uses the default Kafka partitioner to partition data. fixed: Each Flink partition corresponds to a fixed Kafka partition. round-robin: Data in a Flink partition is distributed to the Kafka partitions in round-robin sequence. Custom partition mapping pattern: You can create a subclass of FlinkKafkaPartitioner to configure a custom partition mapping pattern, such as org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Semantic pattern for the Kafka sink table.	String	No	at-least-once	Valid values: none: The delivery semantics is not ensured. Data may be lost or duplicated. at-least-once (default): Ensures that data is not lost. However, data may be duplicated. exactly-once: Kafka transactions are used to ensure the exactly-once semantics. This ensures that data is not lost or duplicated. Note You must configure the sink.transactional-id-prefix option if you set this option to exactly-once.
  sink.transactional-id-prefix	The prefix of the Kafka transaction ID that is used in the exactly-once semantics.	String	No	None	This option takes effect only when the sink.delivery-guarantee option is set to exactly-once.
  sink.parallelism	The parallelism of operators for the Kafka sink table.	Integer	No	None	The parallelism of upstream operators, which 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',
  /*Configure Secure Sockets Layer (SSL).*/
  /*Specify the path of the CA certificate truststore provided by the server.*/
  /*Uploaded artifacts are stored in /flink/usrlib/.*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*Specify the path of the private key file keystore if client authentication is required.*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*The algorithm used by the client to verify the server address. A null value indicates that server address verification is disabled.*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*Configure SASL.*/
  /*Configure SCRAM-SHA-256 as the SASL mechanism.*/
  '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 use the Artifacts feature in the development console to upload the CA certificate and private key from the example. After the upload, the file is stored in the /flink/usrlib directory. If the CA certificate file that you want to use is named my-truststore.jks, you can set the 'properties.ssl.truststore.location' parameter in the WITH clause in the following two ways to use this certificate:

• If you set 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks', Flink does not need to dynamically download OSS files during runtime, but debug mode is not supported.

• For real-time computing engine versions Ververica Runtime (VVR) 11.5 and later, you can configure properties.ssl.truststore.location and properties.ssl.keystore.location with an OSS absolute path. The file path format is oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<project name>/<file name>. This method dynamically downloads the OSS files during Flink runtime and supports debug mode.

Start offset for a Kafka source table

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume data from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume data from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume data from the offset that is committed by 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, the consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If my-group is used for the first time, the consumption starts from the latest offset.
  -- Consume data from the timestamp 1655395200000, in milliseconds.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume data from the specified offset.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Source Table Offset Submission

The Kafka source table commits a consumer offset to the Kafka cluster only after the checkpointing operation succeeds. If the checkpoint interval that you specify is excessively large, the consumer offset is committed with a delay to the Kafka cluster. During the checkpointing operation, the Kafka source table stores the current data reading progress in the state backend. The offset that is committed to the Kafka cluster is not used for fault recovery. The committed offset is used only to monitor the data reading progress in Kafka. Data accuracy is not affected even if the offset fails to be committed.

Custom partitioner for a sink table

If the built-in Kafka producer partitioner does not meet your requirements, you can implement a custom partitioner to write data to specific partitions. A custom partitioner must inherit FlinkKafkaPartitioner. After development, compile the JAR package and use the File Management feature to upload it to the Real-time Computing console. After the JAR package is uploaded and referenced, set the sink.partitioner parameter in the WITH clause. The parameter value must be the full class path of the partitioner, such as org.mycompany.MyPartitioner.

Comparison among Kafka, Upsert Kafka, and Kafka JSON catalogs

Kafka is a message queue system that supports only data insertion and does not support updates or deletions. Therefore, Kafka cannot process Change Data Capture (CDC) data from upstream systems or retraction logic from operators such as aggregate and join during streaming SQL computation. If you want to write data containing change data or retraction data to Kafka, use an Upsert Kafka sink table, which performs special processing on change data.

If you want to synchronize change data from one or more data tables in an upstream database to Kafka in batches, you can use a Kafka JSON catalog. If data stored in Kafka is in JSON format, you can use a Kafka JSON catalog. This eliminates the need to configure the schema and options in the WITH clause. For more information, see Manage Kafka JSON catalogs.

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. Expand 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

• Create a Kafka source

  The Kafka source provides a builder class for creating an instance of KafkaSource. The following sample code shows how to construct a Kafka source to consume messages from the earliest offset of the "input-topic" topic, with the consumer group named my-group, and deserialize the Kafka message body 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 create a KafkaSource, you must specify the following parameters.

  Parameter	Description
  BootstrapServers	The addresses of Kafka brokers. You can call the setBootstrapServers(String) operation to configure the addresses.
  GroupId	The ID of the consumer group. You can call the setGroupId(String) method to configure the ID.
  Topics or Partition	The topics or names of the partitions to which you subscribe. You can configure a Kafka source to subscribe to topics or partitions using one of the following subscription patterns: Topic list. After you configure a topic list, the Kafka source subscribes to all partitions of the specified topics. KafkaSource.builder().setTopics("topic-a","topic-b") Topic pattern. After you specify a regular expression, the Kafka source subscribes to all partitions of the topics that match the specified regular expression. KafkaSource.builder().setTopicPattern("topic.*") Partition list. After you configure a partition list, the Kafka source 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	A deserializer that deserializes Kafka messages. You can call the setDeserializer(KafkaRecordDeserializationSchema) method to specify a deserializer. The KafkaRecordDeserializationSchema interface defines how a ConsumerRecord object is deserialized. You can use one of the following methods to deserialize only the Value fields in the Kafka messages of the ConsumerRecord object: A Kafka source provides the setValueOnlyDeserializer(DeserializationSchema) method. The DeserializationSchema class defines how a Kafka message that is stored as a binary value is deserialized. Use the classes that implement the Deserializer interface of Kafka. For example, you can use the StringDeserializer class to deserialize a message into a string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Note If you want to deserialize a ConsumerRecord object, you must create a class that implements the KafkaRecordDeserializationSchema interface.

  XML

  The Kafka DataStream connectors are stored in the Maven central repository.

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

  When you use a Kafka DataStream connector, you must familiarize yourself with the following Kafka properties:

  • Start offset

    You can use an offset initializer to specify an offset for a Kafka source when the Kafka source starts to read data. An offset initializer is an object that implements the OffsetsInitializer interface. The KafkaSource class provides the following built-in offset initializers.

    Offset initializer	Code Settings
    Reads data from the earliest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Reads data from the latest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Start consuming data with a timestamp greater than or equal to the specified time (in milliseconds).	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Consume from the offset committed by the consumer group. If no such offset exists, use the earliest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Reads data from the committed offset of each partition and no reset strategy is specified.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Note

    • If the built-in offset initializers do not meet your business requirements, you can create custom offset initializers.

    • If you do not specify an offset initializer, the OffsetsInitializer.earliest() offset initializer is used by default.

  • Streaming execution mode and batch execution mode

    A Kafka source can operate in streaming mode or batch mode. By default, a Kafka source operates in streaming mode. In this mode, the deployment continues to run until the deployment fails or is canceled. If you want a Kafka source to operate in batch mode, you can call the setBounded(OffsetsInitializer) method to specify a stop offset. When all partitions reach their stop offsets, the Kafka source exits.

    Note

    In most cases, a Kafka source that operates in streaming mode does not have a stop offset. If you want to debug a Kafka source that operates in streaming mode, you can call the setUnbounded(OffsetsInitializer) method to specify a stop offset. The methods that you can use to specify a stop offset vary based on whether you use the streaming mode or batch mode.

  • Dynamic partition discovery

    If you want a running deployment to process data from new topics and new partitions that match your subscription pattern without restarting the deployment, you can enable the dynamic partition discovery feature on the Kafka source. In the DataStream connector, this feature is disabled by default and must be manually enabled:

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

    Important

    The dynamic partition discovery feature depends on the Kafka cluster's metadata update mechanism. If the Kafka cluster does not promptly update partition information, new partitions might not be discovered. Ensure that the Kafka cluster's partition.discovery.interval.ms configuration matches the actual situation.

  • Event time and watermarks

    By default, a Kafka source uses the timestamp attached to a record as the event time for the record. You can define a watermark strategy based on the event time of each record and send the watermarks to downstream services.

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

    For more information about how to define a custom watermark strategy, see Generating Watermarks.

    Note

    If some source subtasks remain idle for a prolonged period of time, such as a Kafka partition receiving no new messages or the source parallelism exceeding the Kafka partition count, watermark generation may fail. In this case, window computations cannot be triggered, and data processing will stop.

    The solutions are as follows:

    • Configure the watermark timeout mechanism: Enable the table.exec.source.idle-timeout parameter to force the system to generate watermarks after the specified timeout period, ensuring the progress of window computation epochs.

    • Optimize the data source: Set the source parallelism to be equal to or less than the number of Kafka partitions.

  • Consumer Offset Commit

    When a checkpoint is generated, a Kafka source commits the Kafka consumer offset of each partition to Kafka brokers. This ensures that the Kafka consumer offsets recorded on Kafka brokers are consistent with the state of the checkpoint. The Kafka consumer can automatically commit the offsets on each partition to Kafka brokers on a regular basis. You can configure the automatic offset commit feature using the enable.auto.commit and auto.commit.interval.ms options. If you disable the checkpointing feature, a Kafka source relies on the Kafka consumer to commit the offsets to Kafka brokers.

    Note

    Kafka sources do not use the committed offsets recorded on Kafka brokers for fault tolerance. When you commit offsets, Kafka brokers can monitor the progress of record consumption on each partition.

  • Additional properties

    You can call the setProperties(Properties) and setProperty(String, String) methods to configure additional properties for the Kafka source and Kafka consumer. The following table describes the properties of a Kafka source.

    Configuration item	Description
    client.id.prefix	Specifies the prefix for the client ID of the Kafka consumer.
    partition.discovery.interval.ms	Specifies the time interval at which the Kafka source checks for new partitions. Note The partition.discovery.interval.ms property is overridden to -1 in batch mode.
    register.consumer.metrics	Specifies whether to register metrics for the Kafka consumer in Realtime Compute for Apache Flink.
    Other Kafka Consumer configurations	For more information about the properties of a Kafka consumer, see Apache Kafka. Important The Kafka DataStream connector overwrites the values of the following properties: key.deserializer: The value of this property is set to ByteArrayDeserializer. value.deserializer: The value of this property is set to ByteArrayDeserializer. auto.offset.reset.strategy: The value of this property is set to OffsetsInitializer#getAutoOffsetResetStrategy().

    The following sample code shows how the Kafka consumer connects to the Kafka cluster using a JAAS configuration and the SASL/PLAIN authentication mechanism.

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

  • Monitoring

    Kafka sources register metrics in Realtime Compute for Apache Flink for monitoring and diagnosis.

    • Metric scope

      All metrics of a Kafka source are registered under the KafkaSourceReader metric group. KafkaSourceReader is a subgroup of the operator metric group. The metrics for a specific partition are registered in the KafkaSourceReader.topic.<topic_name>.partition.<partition_id> metric group.

      For example, a topic is named my-topic and the partition of the topic is named 1. The consumer offset of the partition is reported by the <some_parent_groups>.operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset metric. The number of successful commits of consumer offsets is measured by the <some_parent_groups>.operator.KafkaSourceReader.commitsSucceeded metric.

    • Metrics

      Metric	Description	Scope
      currentOffset	Current Consumer Offset	TopicPartition
      committedOffset	Current commit offset	TopicPartition
      commitsSucceeded	Number of successful submissions	KafkaSourceReader
      commitsFailed	Failed Submission Count	KafkaSourceReader

    • Kafka Consumer metrics

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

      You can configure the register.consumer.metrics option to specify whether to register metrics for the Kafka consumer. By default, the register.consumer.metrics option is set to true. For more information about the metrics for a Kafka consumer, see Apache Kafka.

• Create a Kafka sink

  A Kafka sink can write data from multiple streams 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 must configure the following parameters.

  Parameter	Description
  Topic	The name of the topic to which data is written.
  Data serialization	When you build a Kafka sink, you must provide a KafkaRecordSerializationSchema to convert input data into a Kafka ProducerRecord object. Flink provides a schema builder that offers common components, such as message key and value serialization, topic selection, and message partitioning. You can also implement the corresponding interfaces for more advanced control. A Kafka sink calls the ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) method for each incoming record to generate a ProducerRecord object that represents the serialized record. Then, the Kafka sink writes the ProducerRecord object to the required topic. You can control how each record is written to Kafka in detail. Using ProducerRecord, you can perform the following actions: Set the name of the target topic. Define the message key. Specify the target partition.
  Kafka client properties	The bootstrap.servers property is required. Specify a comma-separated list of Kafka broker addresses.
  Fault tolerance semantics	After you enable the checkpointing feature, a Kafka sink can ensure exactly-once delivery. You can also configure the DeliveryGuarantee parameter to specify different fault tolerance semantics. Details about the DeliveryGuarantee parameter are as follows: DeliveryGuarantee.NONE: No delivery guarantees are provided by Flink. Data may be lost or duplicated. DeliveryGuarantee.AT_LEAST_ONCE: The Kafka sink ensures that data is not lost. However, data may be duplicated. DeliveryGuarantee.EXACTLY_ONCE: The Kafka sink ensures that data is not lost or duplicated. The Kafka transaction mechanism is used to ensure exactly-once delivery. Note For more information about exactly-once semantics, see Usage notes of Semantic.EXACTLY_ONCE.

Dirty data tolerance and collection

In some cases, your Kafka data source may contain malformed data (dirty data). To prevent frequent job restarts due to such dirty data, you can configure the job to ignore these exceptions. Sample configuration:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable dirty data tolerance.
  ingestion.ignore-errors: true
  # Tolerate up to 1000 dirty data records.
  ingestion.error-tolerance.max-count: 1000

This configuration tolerates up to 1000 dirty data records, allowing your job to run normally when small amounts of dirty data exist. When the number of dirty data records exceeds this threshold, the job fails, prompting you to validate your data.

To ensure your job never fails due to dirty data, use the following configuration:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable dirty data tolerance.
  ingestion.ignore-errors: true
  # Tolerate all dirty data.
  ingestion.error-tolerance.max-count: -1

The dirty data toleration policy prevents jobs from failing frequently due to abnormal data. You may also want to learn more about dirty data to adjust the behavior of Kafka producers. For the process described in Dirty Data Collection, you can view the dirty data of your job in the TaskManager logs. The following code shows a configuration example:

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Enable dirty data tolerance.
  ingestion.ignore-errors: true
  # Tolerate all dirty data.
  ingestion.error-tolerance.max-count: -1

pipeline:
  dirty-data.collector:
    # Write dirty data to TaskManager log files.
    type: logger

Table name and topic mapping strategy

When using Kafka as a data ingestion sink, the message format (debezium-json or canal-json) often includes table name information. Consumers typically use this table name—not the topic name—as the actual table name. Therefore, carefully configure the table name and topic mapping strategy.

Assume you need to synchronize two tables—mydb.mytable1 and mydb.mytable2—from MySQL. Possible mapping strategies include the following:

1. Do not configure any mapping strategy

Without any mapping strategy, each table is written to a topic named after the database and table (e.g., mydb.mytable1). Thus, mydb.mytable1 data goes to the mydb.mytable1 topic, and mydb.mytable2 data goes to the mydb.mytable2 topic. Sample configuration:

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 topics named after the database and table. Instead, they configure route rules to map data to specific topics. Sample configuration:

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, modifying the topic name via route rules also modifies the table name in Kafka messages (in debezium-json or canal-json format). As a result, all messages in this topic have the table name mytable1. Other systems consuming this topic may behave unexpectedly.

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

To retain the original table name information while mapping to a custom topic, use the sink.tableId-to-topic.mapping parameter. Sample configurations:

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 configuration, all data from `mydb.mytable1` and `mydb.mytable2` is written to the `mytable1` topic, but the original table name (`mydb.mytable1` or `mydb.mytable2`) is preserved within the Kafka message format (`debezium-json` or `canal-json`). This allows other systems that consume messages from this topic to correctly retrieve the source table name information.

EXACTLY_ONCE semantic considerations

• Configure consumer isolation level

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

  • read_committed: Reads only committed data.

  • read_uncommitted (default): Lets you read uncommitted data.

  EXACTLY_ONCE relies on read_committed. Otherwise, consumers may see uncommitted data, which breaks consistency.

• Transaction timeout and data loss

  When recovering from a checkpoint, Flink depends only on transactions committed before the checkpoint begins. If the time between job crash and restart exceeds the Kafka transaction timeout, Kafka automatically aborts the transaction, causing data loss.

  • The Kafka broker default transaction.max.timeout.ms = 15 minutes.

  • The Flink Kafka sink sets transaction.timeout.ms to 1 hour by default.

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

• Kafka producer pool and concurrent checkpoints

  The EXACTLY_ONCE mode uses a fixed-size Kafka producer pool. Each checkpoint consumes one producer from the pool. If concurrent checkpoints exceed the pool size, the job fails.

  Adjust the producer pool size according to your maximum concurrent checkpoint count.

• Scale-in restrictions

  If a job fails before the first checkpoint, the producer pool information is not retained after restart. Therefore, before the first checkpoint completes, do not scale down the job's parallelism. If you must scale down, the parallelism must not be less than FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transactional blocking of reads

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

  For example:

  • Transaction 1 writes data.

  • Transaction 2 writes and commits data.

  • Until Transaction 1 completes, Transaction 2's data is invisible to consumers.

  Therefore:

  • During normal operation, data visibility is delayed by approximately the average checkpoint interval.

  • Upon job failure, the topic being written to blocks consumers until the job restarts or the transaction times out. In extreme cases, transaction timeouts can even affect reads.

FAQ

• How do I obtain JSON data in Kafka using Realtime Compute for Apache Flink?

• Why does no data output occur after Kafka source table data is processed using event time-based windows?

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

• How do I resolve field naming conflicts?

• Storage management and operations