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: host1:port1,host2:port2,.... 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	–	Examples: user_event_.*: Matches all topics prefixed with user_event_. prod\.logs\..*: Matches topics prefixed with prod.logs. (the . character must be escaped). 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 the 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 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'
);

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