Realtime Compute for Apache Flink:ApsaraMQ for Kafka

Background information

Prerequisites

Precautions

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

• Each checkpoint generates a transaction ID. If the checkpoint interval is too short, too many transaction IDs are generated. The coordinator of the Kafka cluster may run out of memory, which can 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, which can 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. If one job fails to write, it blocks the Log Start Offset (LSO) of the Kafka partition from advancing. This affects all consumers that read data from that partition.

If you require exactly-once semantics, use Upsert Kafka to write to a primary key table and ensure idempotence with the primary key. If you must use transactional writes, see Considerations for EXACTLY_ONCE semantics.

Troubleshoot network connectivity

If a Flink job reports the error Timed out waiting for a node assignment during startup, the cause is usually a network connectivity issue between Flink and Kafka.

SQL

The Kafka connector can be used 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 as the timestamp of the 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 table type.	String	Yes	None	The value is fixed to Kafka.
  properties.bootstrap.servers	The Kafka broker addresses.	String	Yes	None	The format is host:port,host:port,host:port. Separate addresses with commas (,).
  properties.*	Direct configurations for the Kafka client.	String	No	None	The suffix must be a configuration defined in the official Kafka documentation for producers and consumers. 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 parameter settings, 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	Separate multiple field names with semicolons (;). For example: field1;field2
  key.fields-prefix	Specifies a custom prefix for all key fields of Kafka messages to avoid name conflicts with the fields in the value part of the message.	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 Kafka messages. 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 set only 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	Separate multiple topic names with semicolons (;), for example, topic-1;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 is running.	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	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 discovering 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 Ververica Runtime (VVR) 6.0.x, dynamic partition discovery is disabled by default. Starting from VVR 8.0, this feature is enabled by default, with the discovery interval set to 5 minutes.
  scan.header-filter	Filters data based on whether the Kafka data contains a specified header.	String	No	None	Separate the header key and value with a colon (:). Connect multiple header conditions with logical operators (&, |). The NOT logical operator (!) is also supported. For example, depart:toy|depart:book&!env:test retains Kafka data whose header contains depart=toy or depart=book, and does not contain env=test. Note This parameter is supported only in Ververica Runtime (VVR) 8.0.6 and later. Parentheses are not supported in logical operations. 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, preventing 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 partitioner. 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 manner. Custom partitioner: If fixed and round-robin do not meet your needs, you can create a child class of FlinkKafkaPartitioner to define a custom partitioner. For example: org.mycompany.MyPartitioner
  sink.delivery-guarantee	The delivery semantics for the Kafka sink table.	String	No	at-least-once	Valid values: none: No guarantee. 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 Kafka transaction IDs 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 concurrency 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 to 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 to 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*/
  /*Set the SASL mechanism to 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 uploading, the files are stored in the /flink/usrlib directory. If the CA certificate file to be used is named my-truststore.jks, specify 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' in the WITH parameters to use this certificate.

Source table start offset

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Start consuming from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Start consuming from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Start consuming 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, start consuming from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, start consuming from the latest offset.
  -- Start consuming from the specified millisecond timestamp 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Start consuming 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 lag. 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 data correctness.

Custom partitioner for sink tables

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

Choosing between Kafka, Upsert Kafka, and Kafka JSON catalog

Kafka is an append-only message queue system that does not support data updates or deletions. Therefore, it cannot handle upstream Change Data Capture (CDC) data or retraction logic from operators such as aggregations and joins in streaming SQL. To write data with changes or retractions to Kafka, use the Upsert Kafka sink table, which is specially designed to handle change data.

To conveniently 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 schemas and WITH parameters. For more information, see Manage a Kafka JSON catalog.

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 a KafkaSource instance. The following code example shows how to build a Kafka source to consume data from the earliest offset of `input-topic`. The consumer group is named `my-group`, and the Kafka message body is deserialized into 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 building a KafkaSource, you must specify the following parameters.

  Parameter	Description
  BootstrapServers	The Kafka broker addresses. Configure this using the setBootstrapServers(String) method.
  GroupId	The consumer group ID. Configure this using the setGroupId(String) method.
  Topics or Partition	The names of the subscribed topics or partitions. 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 of 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 a Kafka 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 using the Kafka DataStream connector, you need to understand the following Kafka properties:

  • Start consumer offset

    The Kafka source can specify the starting offset for consumption through an offset initializer (OffsetsInitializer). The built-in offset initializers include the following.

    Offset initializer	Code setting
    Start consuming from the earliest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Start consuming from the latest offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Start consuming from data with a timestamp 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 strategy.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Note

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

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

  • Streaming and batch modes

    The Kafka source supports both streaming and batch runtime 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, there is no stop offset in streaming mode. To facilitate code debugging, you can use setUnbounded(OffsetsInitializer) to specify a stop offset in streaming mode. Note that the method names for specifying stop offsets in streaming and batch modes (setUnbounded and setBounded) are different.

  • Dynamic partition discovery

    To handle scenarios such as topic scaling 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

    The dynamic partition discovery feature 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 promptly, 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 define a custom 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 if the source concurrency 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. This ensures that window calculation cycles progress.

    • Optimize the data source: We recommend maintaining a reasonable ratio of Kafka partitions to source concurrency (recommended: number of partitions ≥ source degree of parallelism).

  • Consumer offset commit

    The Kafka source commits the current consumer offset when a checkpoint is completed. This ensures that the Flink checkpoint state is consistent with 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 a failed job. Committing the offset is only for reporting the consumption progress of the Kafka consumer and the consumer group for monitoring on the broker side.

  • Other properties

    In addition to the properties mentioned above, you can 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 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 diagnostics.

    • 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 the 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 metrics of the Kafka consumer 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 metrics of the Kafka consumer. 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 which data is written.
  Data serialization	When building, you need to provide a KafkaRecordSerializationSchema to convert the input data into a Kafka ProducerRecord. Flink provides a schema builder to offer some 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 that flows in 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 which data is written.
  Kafka client properties	bootstrap.servers is required. It is a comma-separated list of Kafka brokers.
  Fault tolerance semantics	When Flink's checkpointing is enabled, the Flink Kafka sink can guarantee exactly-once semantics. In addition to enabling Flink's checkpointing, you can also specify different fault tolerance semantics through the DeliveryGuarantee parameter. The DeliveryGuarantee parameter details are as follows: DeliveryGuarantee.NONE: (Default setting) Flink makes no guarantees. 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 considerations when using EXACTLY_ONCE semantics, see Considerations for EXACTLY_ONCE semantics.

Table name to topic mapping strategy

When using Kafka as the target for a data integration job, you need to configure the table name to topic mapping strategy carefully. This is because the written Kafka message format (Debezium JSON or Canal JSON) also contains table name information, and subsequent consumption of Kafka messages often uses the table name information in the data as the actual table name (rather than 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 in the format "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 written topic to be in the "database_name.table_name" format. They want to write data to a specified topic. Therefore, 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 route rules to change the written topic name, it also modifies 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`. This may lead to 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 table name to topic mapping rule while preserving 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 obtain the source table name information.

Considerations for EXACTLY_ONCE semantics

• Configure the 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 might see uncommitted data, which breaks consistency.

• Transaction timeout and data loss

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

  • The Kafka broker's default transaction.max.timeout.ms is 15 minutes.

  • The Flink Kafka sink's default transaction.timeout.ms is 1 hour.

  • You must increase transaction.max.timeout.ms on the broker side to be no less than Flink's setting.

• Producer pool and concurrent checkpoints

  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.

• Limitations on scaling in the degree of parallelism

  If a 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 not be lower than FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transaction blocking reads

  In read_committed mode, any open 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 reads.

FAQ

• How does Flink retrieve JSON data from Kafka?

• Why does a Kafka source table with an event time-based window 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