Realtime Compute for Apache Flink:Kafka connector

Overview

Prerequisites

Notes

Transactional writes are not recommended due to known design limitations in Apache Flink and Apache Kafka. When you set sink.delivery-guarantee = 'exactly-once', the Kafka connector enables transactional writes, with the following known issues:

• Each checkpoint generates a new Transaction ID. If the checkpoint interval is too short, the resulting flood of Transaction IDs can cause the Kafka cluster coordinator to run out of memory, compromising cluster stability.

• Each transaction creates a new Producer instance. If too many transactions commit concurrently, the TaskManager can run out of memory, destabilizing the Apache Flink job.

• If multiple Apache Flink jobs use the same sink.transactional-id-prefix, their generated Transaction IDs can conflict. When a write operation fails in one job, it can prevent the Log Start Offset (LSO) of an Apache Kafka partition from advancing. This affects all consumers of that partition.

If you require exactly-once semantics, use the Upsert Kafka connector to write to a primary key table, ensuring idempotence. If you must use transactional writes, see Exactly-once semantics usage notes.

Troubleshoot network connectivity

A Timed out waiting for a node assignment error when a Realtime Compute for Apache Flink job fails to start typically indicates a network connectivity issue between Realtime Compute for Apache Flink and the Kafka cluster.

SQL

Use the Kafka connector as a source table or a sink table in SQL jobs.

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 message topic as the `record_topic` column
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` column
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the message offset as the `record_offset` column
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

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

Connector options

• General

  Option	Description	Type	Required	Default	Remarks
  connector	The connector type.	String	Yes	–	The value must be kafka.
  properties.bootstrap.servers	A list of Kafka broker addresses.	String	Yes	–	Format: host:port,host:port,.... Separate addresses with commas (,).
  properties.*	Additional properties for the Kafka client.	String	No	–	The property keys must be valid options defined in the official Apache Kafka documentation for Producer Configs and Consumer Configs. Realtime Compute for Apache Flink removes the properties. prefix and passes the remaining key-value pairs to the underlying Kafka client. For example, you can set 'properties.allow.auto.create.topics' = 'false' to disable automatic topic creation. The Kafka connector overwrites these options, so you cannot configure them this way: key.deserializer value.deserializer
  format	The format for serializing and deserializing the value of a Kafka message.	String	No	–	Supported formats: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Note For more information, see Format options.
  key.format	The format for serializing and deserializing the key of a Kafka message.	String	No	–	Supported formats: csv json avro debezium-json canal-json maxwell-json avro-confluent raw Note When you use this configuration, key.options is required.
  key.fields	The fields from the table schema to use as the Kafka message key.	String	No	–	Separate multiple field names with semicolons (;). For example, 'field1;field2'.
  key.fields-prefix	A custom prefix for all key fields to prevent name conflicts with value fields.	String	No	–	This prefix is used to distinguish between key and value fields. It is removed before serializing the key or after deserializing it. Note If you use this option, value.fields-include must be set to EXCEPT_KEY.
  value.format	The format for serializing and deserializing the value of a Kafka message.	String	No	–	This configuration is equivalent to format. You can set only one of format or value.format. If both are configured, value.format overrides format.
  value.fields-include	Defines whether the key fields are included in the value format.	String	No	ALL	Valid values: ALL: The Kafka message value includes all table columns. EXCEPT_KEY: The Kafka message value includes all table columns except those defined in key.fields.

• Source table

  Option	Description	Type	Required	Default	Remarks
  topic	The topic or topics to read from.	String	No	–	To subscribe to multiple topics, separate their names with semicolons (;), for example, 'topic-1;topic-2'. Note You can specify either this option or topic-pattern, but not both.
  topic-pattern	A regular expression that matches the topics to subscribe to. The consumer subscribes to all topics whose names match this pattern.	String	No	–	Note You can specify either this option or topic, but not both.
  properties.group.id	Kafka source consumer group ID.	String	No	KafkaSource-{Source-Table-Name}	If you use a consumer group ID for the first time, you must also set properties.auto.offset.reset to either earliest or latest to define the initial startup offset.
  scan.startup.mode	Kafka consumer startup offset.	String	No	group-offsets	Valid values: earliest-offset: Starts reading from the earliest available offset. latest-offset: Starts reading from the latest offset. group-offsets: Starts reading from the committed offsets of the specified properties.group.id. timestamp: Starts reading from the specified scan.startup.timestamp-millis. specific-offsets: Starts reading from the offsets specified in scan.startup.specific-offsets. Note This option applies only when a job starts without state. If a job resumes from a checkpoint, it reads from the offsets stored in the checkpoint state.
  scan.startup.specific-offsets	Per-partition start offset when scan.startup.mode is specific-offsets.	String	No	–	For example, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	Start timestamp in milliseconds when scan.startup.mode is set to timestamp.	Long	No	–	The unit is milliseconds.
  scan.topic-partition-discovery.interval	Partition discovery interval.	Duration	No	5 minutes	The connector periodically discovers and reads from new partitions. When you use topic-pattern, the connector also discovers new topics that match the pattern. Set the interval to a non-positive value to disable this feature. 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 a discovery interval of 5 minutes.
  scan.header-filter	Filters messages based on Kafka message headers.	String	No	–	A header key and its value are separated by a colon (:). Multiple header conditions are connected by using logical operators (& and |). The NOT logical operator (!) is also supported. For example, depart:toy|depart:book&!env:test retains Kafka data if the header contains depart=toy or depart=book and does not contain env=test. Note This option is supported only in Ververica Runtime (VVR) 8.0.6 and later. Parentheses in expressions are not supported. Logical operations are evaluated left to right. Header values are converted to UTF-8 strings for comparison.
  scan.check.duplicated.group.id	Checks whether another active consumer is already using the properties.group.id.	Boolean	No	false	Valid values: true: Before starting the job, the system checks for a duplicate consumer group. If one is found, the job fails to prevent conflicts. false: Starts the job without checking for conflicts. Note This option is supported only in Ververica Runtime (VVR) 6.0.4 and later.

• Sink table

  Option	Description	Type	Required	Default	Remarks
  topic	Target topic.	String	Yes	–	–
  sink.partitioner	Maps records from parallel sink instances to Kafka partitions.	String	No	default	Valid values: default: Uses the default Kafka partitioner. fixed: Each parallel sink instance writes to a fixed Kafka partition. round-robin: Records are distributed to partitions in a round-robin fashion. Custom partitioner: To use a custom partitioner, provide the fully qualified class name of a FlinkKafkaPartitioner subclass, for example, org.mycompany.MyPartitioner.
  sink.delivery-guarantee	Sink delivery guarantee.	String	No	at-least-once	Valid values: none: Provides no guarantees. Records may be lost or duplicated. at-least-once: Guarantees that no records are lost, but they might be duplicated. exactly-once: Uses Kafka transactions to provide exactly-once semantics, ensuring records are neither lost nor duplicated. Note When using exactly-once semantics, you must also specify sink.transactional-id-prefix.
  sink.transactional-id-prefix	Transaction ID prefix. Required when sink.delivery-guarantee is exactly-once.	String	Yes, if sink.delivery-guarantee is exactly-once	–	Required only when sink.delivery-guarantee is set to exactly-once.
  sink.parallelism	Sink operator parallelism.	Integer	No	–	By default, the framework determines the parallelism based on the upstream operators.

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 */
  /* Path to the truststore for the server's CA certificate. */
  /* Files uploaded using Artifacts are stored in the /flink/usrlib/ directory. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* If client authentication is required, you must also 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 used to verify the server hostname. An empty string disables hostname 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 use the Artifacts feature of the Realtime Compute for Apache Flink console to upload the CA certificate and private key mentioned in the example. The uploaded files are stored in the /flink/usrlib directory. To use a CA certificate file named my-truststore.jks, you can set the 'properties.ssl.truststore.location' property in the WITH clause in one of the following two ways:

• Set 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks'. This method avoids dynamically downloading files from Object Storage Service (OSS) at runtime, but it does not support Debug Mode.

• If the Realtime Compute engine version is VVR 11.5 or later, you can configure properties.ssl.truststore.location and properties.ssl.keystore.location to an absolute OSS path. The file path format is oss://flink-fullymanaged-<Workspace ID>/artifacts/namespaces/<Namespace name>/<file name>. This method dynamically downloads the OSS files during Flink runtime and supports Debug Mode.

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 offsets 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, consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consumption starts from the latest offset.
  -- Consume from the specified timestamp in milliseconds: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from specific offsets.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

Committing source offsets

The Kafka source table commits its consumer offset to the Kafka cluster only after a successful checkpoint, so a long checkpoint interval causes the committed offset to lag. The source table stores its actual reading progress in checkpoint state, which the system uses for fault recovery. Committed offsets serve only as a progress monitor and are not used for recovery, so commit failures do not affect data accuracy.

Custom sink partitioner

If Kafka's built-in partitioning strategy does not meet your requirements, you can implement a custom partitioner by extending the FlinkKafkaPartitioner class. After development is complete, compile your code into a JAR package and upload it using the Artifacts feature in the Realtime Compute console. After the JAR package is uploaded and referenced, set the sink.partitioner parameter in the WITH clause to your partitioner's fully qualified class name, for example, org.mycompany.MyPartitioner.

Kafka, Upsert Kafka, and Kafka JSON catalog

Kafka is an append-only event streaming platform that does not support data updates or deletions. In streaming SQL, a standard Kafka sink table cannot handle upstream Change Data Capture (CDC) data or the retraction logic of operators such as aggregate and join. If you need to write data that contains changes or retractions, use an Realtime Compute for Apache Flink:Upsert Kafka sink table.

To simplify the batch synchronization of Change Data Capture (CDC) data from one or more upstream database tables to Kafka, you can use a Kafka JSON catalog. If the data stored in Kafka is in JSON format, a Kafka JSON catalog lets you skip the step of defining a schema and WITH parameters. For details, 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. 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 Kafka Source instance. The following sample code builds a Kafka Source that consumes data from the earliest Offset of the input-topic Topic. The Consumer Group is my-group, and the Kafka Message Value 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");

  To build a Kafka Source, you must specify the following properties.

  Parameter	Description
  BootstrapServers	A list of Kafka broker addresses. Set this property by calling the setBootstrapServers(String) method.
  GroupId	The ID of the Consumer Group. Set this property by calling the setGroupId(String) method.
  Topics or Partitions	The topics or partitions to subscribe to. The Kafka Source supports the following three methods to subscribe to topics or partitions: Subscribes to all partitions of the topics in a list. KafkaSource.builder().setTopics("topic-a","topic-b") Topic pattern: Subscribe to all partitions of topics whose names match the specified regular expression. KafkaSource.builder().setTopicPattern("topic.*") The list of partitions, where you can subscribe to a specified partition. final HashSet<TopicPartition> partitionSet = new HashSet<>(Arrays.asList( new TopicPartition("topic-a", 0), // Partition 0 of topic "topic-a" new TopicPartition("topic-b", 5))); // Partition 5 of topic "topic-b" KafkaSource.builder().setPartitions(partitionSet)
  Deserializer	The deserializer used to parse Kafka messages. Specify the deserializer using the setDeserializer(KafkaRecordDeserializationSchema) method. KafkaRecordDeserializationSchema defines how to parse a Kafka ConsumerRecord. If you only need to parse the Value of a Kafka Message, you can use one of the following methods: Use the setValueOnlyDeserializer(DeserializationSchema) method from the builder class. The DeserializationSchema defines how to parse the binary data of the Kafka Message Value. Use a class that implements Kafka's Deserializer interface. For example, you can use StringDeserializer to parse the Kafka Message Value into a string. import org.apache.kafka.common.serialization.StringDeserializer; KafkaSource.<String>builder() .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class)); Note To parse a complete ConsumerRecord, you must implement the KafkaRecordDeserializationSchema interface.

  POM

  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, consider the following properties:

  • Starting offset

    A Kafka Source specifies its starting offset using an offset initializer (OffsetsInitializer). The built-in initializers include:

    Offset initializer	Code
    Starts consuming from the earliest Offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Starts consuming from the latest Offset.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Starts consuming data whose timestamp is greater than or equal to the specified time. The unit is milliseconds.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Starts consuming from the committed Offset of the Consumer Group. If no committed Offset exists, it uses the specified reset strategy (for example, earliest Offset).	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Consumption starts from the offset committed by the consumer group, and no offset reset policy is specified.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Note

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

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

  • Streaming mode and batch mode

    The Kafka Source supports both streaming mode and Batch Mode. By default, it operates in streaming mode, where the Job runs indefinitely until it fails or is canceled. To configure the Kafka Source to run in Batch Mode, you can use setBounded(OffsetsInitializer) to specify a stop Offset. The Kafka Source exits when all partitions reach their specified stop offsets.

    Note

    A Kafka Source in streaming mode does not typically have a stop Offset. However, for testing purposes, you can use setUnbounded(OffsetsInitializer) to specify a stop Offset even in streaming mode. Note the different method names for specifying the stop Offset: setUnbounded for streaming mode and setBounded for Batch Mode.

  • Dynamic partition discovery

    To handle Topic scaling or the creation of new topics without restarting the Flink Job, you can enable Dynamic partition discovery when subscribing to topics by pattern. This feature is disabled by default and must be explicitly enabled:

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

    Important

    The dynamic partition discovery feature depends on the metadata update mechanism of the Kafka cluster. If the Kafka cluster does not update partition information in a timely manner, new partitions might not be discovered. Make sure that the partition.discovery.interval.ms configuration of the Kafka cluster matches your actual scenario.

  • Event time and watermark

    By default, the Kafka Source uses the timestamp from the Kafka Message as the Event Time. You can define a custom Watermark strategy to extract the Event Time from the Message body and emit a Watermark downstream.

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

    To learn more about custom Watermark strategies, see Generating Watermarks.

    Note

    If a source subtask is idle (for example, when a Kafka Partition has no new data or the source Parallelism is higher than the number of Kafka partitions), the Watermark for that subtask will not advance. This can block downstream window computations.

    To resolve this issue, consider the following solutions:

    • Configure a source idle timeout: Enable the table.exec.source.idle-timeout Property to mark an idle source as temporarily idle. This allows the downstream Watermark to advance.

    • Set appropriate Parallelism: Ensure the source Parallelism is not greater than the number of Kafka partitions.

  • Offset commit

    When checkpointing is enabled, the Kafka Source commits the current consumer Offset to Kafka when a Checkpoint completes. This ensures the Flink Checkpoint state is consistent with the committed Offset on the Kafka broker. If checkpointing is disabled, the Kafka Source relies on the Kafka consumer's internal automatic periodic Offset commit mechanism. This feature is controlled by the enable.auto.commit and auto.commit.interval.ms Kafka consumer properties.

    Note

    The Kafka Source does not rely on committed offsets for fault tolerance and recovery. Committing offsets is only for monitoring the progress of the Kafka consumer and Consumer Group.

  • Other properties

    In addition to the properties mentioned, you can use setProperties(Properties) and setProperty(String, String) to set any Property for the Kafka Source and its underlying Kafka consumer. The Kafka Source provides the following specific properties.

    Parameter	Description
    client.id.prefix	Client ID prefix for the Kafka consumer.
    partition.discovery.interval.ms	Partition discovery interval in milliseconds. A value of -1 disables dynamic partition discovery. Note In Batch Mode, this property is automatically set to -1.
    register.consumer.metrics	Registers Kafka consumer metrics in Flink.
    Other Kafka Consumer Configuration	For a complete list of Kafka consumer configurations, see the official Apache Kafka documentation. Important To ensure correct operation, the Kafka DataStream Connector overwrites the following manually configured properties: key.deserializer is always overwritten to org.apache.kafka.common.serialization.ByteArrayDeserializer. value.deserializer is always overwritten to org.apache.kafka.common.serialization.ByteArrayDeserializer. auto.offset.reset.strategy is overwritten by the strategy provided by the OffsetsInitializer.

    The following example shows how to configure a Kafka consumer to use the PLAIN 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 exposes metrics through Flink's metrics system for monitoring and diagnostics.

    • Metric scope

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

      For example, the current Consumer Offset metric (currentOffset) for Partition 1 of the "my-topic" topic is available at .operator.KafkaSourceReader.topic.my-topic.partition.1.currentOffset. The number of successful commits (commitsSucceeded) is available at .operator.KafkaSourceReader.commitsSucceeded.

    • List of metrics

      Metric	Description	Scope
      currentOffset	The current consumer Offset of a partition.	TopicPartition
      committedOffset	The last committed Offset for a partition.	TopicPartition
      commitsSucceeded	The total number of successful offset commits.	KafkaSourceReader
      commitsFailed	Number of failed commits	KafkaSourceReader

    • Kafka consumer metrics

      The underlying Kafka consumer's metrics are registered in the KafkaSourceReader.KafkaConsumer metric group. For example, the records-consumed-total Metric is registered at .operator.KafkaSourceReader.KafkaConsumer.records-consumed-total.

      You can use the register.consumer.metrics Property to specify whether to register the Kafka consumer metrics. This option is enabled by default (true). For more information about Kafka consumer metrics, see the Apache Kafka documentation.

• Build a Kafka sink

  The Flink Kafka Sink writes a data stream to one or more Kafka topics.

  DataStream<String> stream = ...
  
  Properties kafkaProperties = new Properties();
  kafkaProperties.setProperty("bootstrap.servers", "localhost:9092");
  
  KafkaSink<String> sink = KafkaSink.<String>builder()
          .setKafkaProducerConfig(kafkaProperties)
          .setRecordSerializer(
                  KafkaRecordSerializationSchema.builder()
                          .setTopic("my-topic")
                          .setValueSerializationSchema(new SimpleStringSchema())
                          .build())
          .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
          .build();
  
  stream.sinkTo(sink);

  To build a Kafka Sink, you must configure the following properties.

  Parameter	Description
  Kafka client properties	The bootstrap.servers Property is required. It specifies a comma-separated list of Kafka brokers.
  Record serializer	You must provide a KafkaRecordSerializationSchema to convert input data into a Kafka ProducerRecord. Flink provides a schema builder that offers common components, such as serialization for message keys and values, topic selection, and message partitioning. You can also implement the corresponding interfaces for more granular control. The ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) method is called for each incoming record to generate a ProducerRecord to be written to Kafka. The ProducerRecord provides fine-grained control over how each record is written to Kafka, allowing you to: Set the destination Topic. Set the Message Key. Specify the destination Partition.
  Delivery guarantee	The bootstrap.servers parameter is required and specifies a comma-separated list of Kafka brokers.
  Delivery guarantee	When Flink checkpoints are enabled, the Flink Kafka Sink can provide exactly-once semantics. In addition to enabling checkpoints, you can use the DeliveryGuarantee parameter to specify different delivery guarantees. The DeliveryGuarantee parameter provides the following options: DeliveryGuarantee.NONE: (Default) Flink provides no guarantees. Data may be lost or duplicated. DeliveryGuarantee.AT_LEAST_ONCE: Guarantees that no data is lost, but duplication may occur. DeliveryGuarantee.EXACTLY_ONCE: Uses Kafka transactions to provide exactly-once semantics. Note When using EXACTLY_ONCE semantics, see Considerations for exactly-once semantics.

Dirty data tolerance and collection

Your Kafka data source may contain malformed records, commonly called dirty data. To keep your job from repeatedly failing and restarting, you can configure it to skip these invalid records. For 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 up to 1000 dirty data records
  ingestion.error-tolerance.max-count: 1000

With this configuration, the job continues running as long as it encounters no more than 1,000 dirty records. Once the count exceeds that threshold, the job fails so you can investigate 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 records
  ingestion.error-tolerance.max-count: -1

While dirty data tolerance keeps your job running, you may also want to inspect the problematic records. You may also want to analyze the dirty data to improve your Kafka producers. As described in Dirty Data Collection, you can view the job's dirty data in the TaskManager logs. For 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 records
  ingestion.error-tolerance.max-count: -1

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

Table name and topic mapping

When Kafka serves as a Flink CDC sink, the message format (such as Debezium JSON or Canal JSON) embeds the original table name. Downstream consumers typically use this embedded name as the table identifier rather than the topic name, so it is important to configure the mapping between table names and topics correctly.

Suppose you need to synchronize two tables from a MySQL database: mydb.mytable1 and mydb.mytable2. The following mapping strategies are available:

1. No mapping strategy

Without any mapping strategy, data for each table is written to a topic named in the <Database Name>.<Table Name> format. 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 is an example 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. Route rule mapping (Not recommended)

You might want to write data to a specific topic instead of using the default <Database Name>.<Table Name> format. To do this, you can configure a route rule. The following is an example 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: mytable

In this case, all data from mydb.mytable1 and mydb.mytable2 is written to a single topic named mytable.

However, a route rule that changes the destination topic also changes the table name in the Kafka message (in Debezium JSON or Canal JSON format). The table name in all Kafka messages becomes mytable. This can cause unexpected behavior in systems that consume messages from this topic.

3. Mapping with sink.tableId-to-topic.mapping (Recommended)

To map table names to topics while preserving the original source table name, use the sink.tableId-to-topic.mapping parameter. The following is an example 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.tableId-to-topic.mapping: mydb.mytable1,mydb.mytable2:mytable

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

Alternatively, you can use the following 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.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 mytable topic, and the table name within the Kafka messages (in Debezium JSON or Canal JSON format) is preserved as mydb.mytable1 or mydb.mytable2. This way, downstream systems can still identify each record's original source table.

Exactly-once semantics

• Configure the consumer isolation level

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

  • 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, breaking consistency.

• Transaction timeout and data loss

  When recovering from a checkpoint, Realtime Compute for Apache Flink considers only transactions that were committed before that checkpoint began. If the duration between a job failure and its restart exceeds the Kafka transaction timeout, Kafka automatically aborts the open transaction, which can result in data loss.

  • The default transaction.max.timeout.ms for a Kafka broker is 15 minutes.

  • By default, Flink Kafka Sink sets the transaction.timeout.ms parameter to 1 hour.

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

• Producer pool and concurrent checkpoints

  The EXACTLY_ONCE mode uses a fixed-size Kafka producer pool. Each checkpoint uses one producer from this pool. If the number of concurrent checkpoints exceeds the pool size, the job fails.

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

• Parallelism scale-down constraints

  If a job fails before the first checkpoint is completed, the original producer pool information is lost on restart. Therefore, do not scale down the job's parallelism before the first checkpoint completes. If a scale-down is necessary, the new parallelism must not be less than FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR.

• Transactions block reads

  In read_committed mode, any transaction that has not been committed or aborted blocks read operations on the entire topic.

  For example:

  • Transaction 1 writes data.

  • Transaction 2 writes more data and is committed.

  • As long as Transaction 1 remains open, the data from the committed Transaction 2 is invisible to consumers.

  This has the following implications:

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

  • If a job fails, any topic it was writing to is blocked for consumers until the job restarts or the transaction times out. In extreme cases, the transaction timeout process itself can also affect read operations.

FAQ

• How do I retrieve JSON data from Kafka using Realtime Compute for Apache Flink?

• Why does an Event Time-based Window produce no output for a ApsaraMQ for Kafka Source Table?

• How do I connect to a secured Kafka Cluster?

• How do I resolve field name conflicts?

• Workspaces and Namespaces