Realtime Compute for Apache Flink:Kafka connector

Background information

Prerequisites

Limits

Limits on the CREATE TABLE AS statement

• Only Realtime Compute for Apache Flink that uses VVR 4.0.12 or later allows you to use Kafka as a data source for the CREATE TABLE AS statement.

• The CREATE TABLE AS statement can infer column data types only from a JSON source and synchronize schema changes of such a table.

• The CREATE TABLE AS statement can infer the data types only for value fields from a Kafka source and synchronize schema changes to those fields. To synchronize data of key fields from a Kafka source, you must specify them in the DDL statement. For more information, see Example 3.

Limits on YAML deployments for data ingestion

• Only Realtime Compute for Apache Flink that uses VVR 8.0.10 or later allows you to use Kafka as a data source for YAML deployments for data ingestion.

• Only the Debezium JSON and Canal JSON formats are supported.

• For a source, data in the same table can be distributed across multiple partitions only in Realtime Compute for Apache Flink that uses VVR 8.0.11.

Usage notes

If using VVR 8.0.11 or earlier to write to Kafka, do not enable transactional writes due to known design flaws in Apache Flink and Apache Kafka. Namely, do not specify 'sink.delivery-guarantee' = 'exactly-once' in the Kafka connector configurations. Transactional writes can result in the following issues:

• Unstable Kafka cluster: Currently, each checkpoint generates a new transaction ID. Short checkpoint intervals can cause an excessive number of transaction IDs to overwhelm the Kafka coordinator, depleting its memory and compromising the stability of the Kafka cluster.

• Unstable Flink job: A Kafka producer is created per transaction. High concurrency in committing transactions can exhaust TaskManager's memory with many transaction ID instances, destabilizing the Flink job.

• Obstructed data consumption: A transaction ID consists of a prefix and a checkpoint ID. If the same prefix (sink.transactional-id-prefix) is shared across Flink jobs, transaction IDs are likely to conflict. And a write exception in a Flink job can prevent the Log Start Offset (LSO) from moving forward in affected Kafka partitions, impacting the data consumption of these partitions.

To deliver data to Kafka exactly once, use the Upsert Kafka connector to create a sink table with a primary key, which can guarantee idempotent writes.

Troubleshoot network connectivity

If the error message Timed out waiting for a node assignment appears when you start a deployment, it indicates a network connectivity problem between your Realtime Compute for Apache Flink workspace and the Kafka cluster.

If the broker is misconfigured, the client will receive an incorrect endpoint. In this case, even if the initial connection can be correctly established via the IP address and port number specified in the properties.bootstrap.servers option, data cannot be read from or written to Kafka. In most cases, this issue occurs if a forwarding mechanism, such as a proxy, a port forwarding, or a leased line, is used to establish the connection between Realtime Compute for Apache Flink and Kafka.

SQL

The Kafka connector can be used in SQL drafts to read data from and write data into Kafka topics.

Syntax

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

Metadata columns

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

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

Connector options in WITH clause

• Common options

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

• Source table

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

• Sink table

  Option	Description	Data type	Required	Default value	Remarks
  topic	The name of the topic to which data is written.	String	Yes	No default value	N/A
  sink.partitioner	The mapping pattern between Realtime Compute for Apache Flink partitions and Kafka partitions.	String	No	default	Valid values: default: The default Kafka partitioner is used to partition data. fixed: Each Flink partition corresponds to a fixed Kafka partition. round-robin: Data in a Flink partition is distributed to the Kafka partitions in round-robin sequence. Custom partition mapping pattern: You can create a subclass of FlinkKafkaPartitioner to configure a custom partition mapping pattern, such as org.mycompany.MyPartitioner.
  sink.delivery-guarantee	The delivery semantics for the Kafka sink table.	String	No	at-least-once	Valid values: none: The delivery semantics is not ensured. Data may be lost or duplicated. at-least-once: It ensures that no data is lost. However, duplicate data may exist. exactly-once: Kafka transactions are used to ensure the exactly-once semantics. This ensures that data is not lost or duplicated. Note You must configure the sink.transactional-id-prefix option if you set this option to exactly-once.
  sink.transactional-id-prefix	The prefix of the Kafka transaction ID that is used in the exactly-once semantics.	String	No	No default value.	This option takes effect only when the sink.delivery-guarantee option is set to exactly-once.
  sink.parallelism	The parallelism of operators for the Kafka sink table.	Integer	No	No default value.	The parallelism of upstream operators, which is determined by the framework.

• CREATE TABLE AS (CTAS) statement exclusive

  Option	Description	Data type	Required	Default value	Remarks
  json.infer-schema.flatten-nested-columns.enable	Specifies whether to recursively expand nested columns in a JSON text.	Boolean	No	false	Valid values: true: Nested columns are recursively expanded. Realtime Compute for Apache Flink uses the path that indexes the value of the column that is expanded as the name of the column. For example, in the JSON {"nested": {"col": true}}, the column col becomes nested.col after expansion. false: Nested types are parsed as the STRING type.
  json.infer-schema.primitive-as-string	Specifies whether to infer all basic types as the STRING type.	Boolean	No	false	Valid values: true: All basic data types are inferred as the STRING type. false: Data types are inferred based on basic rules.

  You can add the properties. prefix to the names of all options that are supported by Kafka consumers and producers and configure the options in the WITH clause. For example, if you want to set the request.timeout.ms option of Kafka consumers or producers to 60000 (measured in milliseconds), you can configure 'properties.request.timeout.ms'='60000' in the WITH clause. For more information about the options supported by Kafka consumers and producers, see Apache Kafka official documentation.

Security and authentication

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /*Configure Secure Sockets Layer (SSL).*/
  /*Specify the path of the CA certificate truststore provided by the server.*/
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /*Specify the path of the private key file keystore if client authentication is required.*/
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /*The algorithm used by the client to verify the server address. A null value indicates that server address verification is disabled.*/
  'properties.ssl.endpoint.identification.algorithm' = '',
  /*Configure SASL.*/
  /*Configure SCRAM-SHA-256 as the SASL mechanism.*/
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /*Configure JAAS.*/
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

To upload the CA certificate and private key, use the Artifacts feature in the development console. For example, if the uploaded file is stored in the /flink/usrlib directory and the name of the CA certificate file that you want to use is my-truststore.jks, you can configure 'properties.ssl.truststore.location' = '/flink/usrlib/my-truststore.jks' in the WITH clause to use the uploaded certificate.

Start offset for a Kafka source table

Startup Mode

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

Priority of start offsets

If one of the preceding offsets becomes invalid due to expiration or an issue in the Kafka cluster, the reset strategy that is specified by the properties.auto.offset.reset option is used to reset the offset. If you do not configure this option, an exception occurs.

In most cases, the Kafka source table starts to read data from the offset that is committed by a consumer group that has a new group ID. When the Kafka source table queries the offset that is committed by the consumer group in the Kafka cluster, no valid offset is returned because the group ID is used for the first time. In this case, the reset strategy that is specified by the properties.auto.offset.reset option is used to reset the offset. Therefore, you must configure the properties.auto.offset.reset option in the scenario when the Kafka source table starts to read data from the offset that is committed by a consumer group that has a new group ID.

Kafka source offset committing

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

Custom partitioner for a sink table

If the built-in Kafka producer cannot meet your business requirements, you can use a custom Kafka partitioner to write data to the related partitions. A custom partitioner needs to inherit FlinkKafkaPartitioner. After you configure a custom partitioner, compile the JAR package for the partitioner. Then, use the Artifacts feature in the development console to upload the JAR. After the JAR is uploaded and referenced, you need to configure the sink.partitioner option in the WITH clause. The value of this option must be a complete classpath of the partitioner, such as org.mycompany.MyPartitioner.

Comparison among Kafka, Upsert Kafka, and a Kafka JSON catalog

Kafka is a message queue system that allows for only data addition and does not support data updates or deletions. Therefore, Kafka cannot process Change Data Capture (CDC) data of upstream systems and the retraction logic of operators such as aggregate and join during streaming SQL computing. If you want to write data that contains change data or retraction data to Kafka, use an Upsert Kafka sink table that can perform special processing on change data.

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

Data synchronization with the CREATE TABLE AS statement

• Type inference

  By default, Realtime Compute for Apache Flink displays only the first layer of data in the JSON text during type inference. Realtime Compute for Apache Flink infers the SQL data types based on the JSON data types, values, and the data type mappings. The following table lists the mappings between JSON data types and Flink SQL data types.

  JSON data type	Flink SQL data type
  BOOLEAN	BOOLEAN
  STRING	DATE, TIMESTAMP, TIMESTAMP_LTZ, TIME, or STRING
  INT or LONG	BIGINT
  BIGINT	DECIMAL or STRING Note The precision of the DECIMAL type in Realtime Compute for Apache Flink is limited. If an integer exceeds the maximum allowed digits for DECIMAL, Realtime Compute for Apache Flink casts it to STRING to avoid precision loss.
  FLOAT, DOUBLE, or BIG DECIMAL	DOUBLE
  ARRAY	STRING
  OBJECT	STRING

  Example

  • JSON text

    {
      "id": 101,
      "name": "VVP",
      "properties": {
        "owner": "Alibaba Cloud",
        "engine": "Flink"
      }
        "type": ["Big data"]
    }

  • Information about the table to which Realtime Compute for Apache Flink writes data

    id	name	properties	type
    101	VVP	{ "owner": "Alibaba Cloud", "engine": "Flink" }	["Big data"]

• Auxiliary type inference

  If type inference based on the preceding rules does not meet your needs, you can explicitly declare the data type of a specific column in the source table's DDL statement. Realtime Compute for Apache Flink prioritize the declared type during parsing. In this example, Realtime Compute for Apache Flink parses the price field as DECIMAL instead of converting it to DOUBLE based on the data type mappings.

  CREATE TABLE evolvingKafkaSource (
    price DECIMAL(18, 2)
  ) WITH (
    'connector' = 'kafka',
    'properties.bootstrap.servers' = 'localhost:9092',
    'topic' = 'evolving_kafka_demo',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
  );

  However, if the data type that you specify in the DDL statement is different from the actual data type, you can use one of the following methods to handle the issue:

  • If a column's declared data type has a wider range than the actual data type, the field is parsed according to the declared data type. For example, if a column declared as the DOUBLE type is used to hold BIGINT data, the column's data type is parsed as DOUBLE.

  • If the actual data type has a wider range than or is incompatible with the declared data type, an error is returned because CREATE TABLE AS does not support type changes. To resolve this, you must restart your deployment and declare a correct data type.

    The following figure shows the ranges of data types and the compatibility between the data types.

    Note

    • In the preceding figure, data types closer to the root node have wider ranges. Data types on different branches are incompatible.

    • Auxiliary type inference is not supported for complex data types, such as ROW, ARRAY, MAP, and MULTISET.

    • By default, Realtime Compute for Apache Flink parses complex data types as STRING.

Examples

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- Optional. Expand all nested columns. 
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
-- Use COALESCE to handle null values.

DataStream API

• Create a Kafka source

  You can use the KafkaSource class to create a Kafka source. The following sample code shows how to create a Kafka source to consume messages from the earliest offset of the "input-topic" topic. The consumer group is my-group. The value of each message 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");

  POM

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

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

  The following table describes the properties that you must configure when you create a Kafka source.

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

  When you use a Kafka DataStream connector, you must be familiar with the following Kafka properties:

  • Start offset

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

    Offset initializer	Code
    Specifies that the Kafka source starts to consume messages from the earliest record of each partition.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.earliest())
    Specifies that the Kafka source starts to consume messages from the latest record of each partition.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.latest())
    Specifies that the Kafka source starts to consume messages from the first record of each partition. The first record has a timestamp that is greater than or equal to the specified timestamp. Unit: milliseconds.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.timestamp(1592323200000L))
    Specifies that the Kafka source starts to consume messages from the committed offset of each partition and a reset strategy is specified. If a partition does not have a committed offset, the reset strategy resets the offset and the Kafka source starts from the earliest record of the partition.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    Specifies that the Kafka source starts to consume messages from the committed offset of each partition and no reset strategy is specified.	KafkaSource.builder().setStartingOffsets(OffsetsInitializer.committedOffsets())

    Note

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

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

  • Streaming execution mode and batch execution mode

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

    Note

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

  • Dynamic partition discovery

    If you want a running deployment to process data from new topics and from new partitions that match your subscription pattern without the need to restart the deployment, you can enable the dynamic partition discovery feature on the Kafka source.

    Note

    By default, the dynamic partition discovery feature is enabled, with an interval of 5 minutes. To disable the dynamic partition discovery feature, you must explicitly set this option to a non-positive value. The following sample code shows how to configure the partition.discovery.interval.ms option:

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

  • Event time and watermarks

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

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

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

  • Consumer offsets

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

    Note

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

  • Additional properties

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

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

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

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

  • Monitoring

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

    • Metric scope

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

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

    • Metrics

      Metric	Description	Scope
      currentOffset	Reports the Kafka consumer offset of a partition.	TopicPartition
      committedOffset	Reports the committed consumer offset of a partition.	TopicPartition
      commitsSucceeded	Reports the number of successful commits of consumer offsets.	KafkaSourceReader
      commitsFailed	Reports the number of failed commits of consumer offsets.	KafkaSourceReader

    • Metrics for a Kafka consumer

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

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

• Build a Kafka sink

  A Kafka sink can write data from multiple streams to one or more Kafka topics.

  DataStream<String> stream = ...
  
  
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", );
  KafkaSink<String> kafkaSink =
                  KafkaSink.<String>builder()
                          .setKafkaProducerConfig(kafkaProperties) // // producer config
                          .setRecordSerializer(
                                  KafkaRecordSerializationSchema.builder()
                                          .setTopic("my-topic") // target topic
                                          .setKafkaValueSerializer(StringSerializer.class) // serialization schema
                                          .build())
                          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // fault-tolerance
                          .build();
  
  stream.sinkTo(kafkaSink);

  The following table describes the required properties:

  Property	Description
  Topic	The name of the topic to which data is written.
  Record serializer	You must provide KafkaRecordSerializationSchema when you build a Kafka sink to convert the input data into a ProducerRecord object of Kafka. Realtime Compute for Apache Flink provides some common components by providing a schema builder, such as message key or message value serialization, topic selection, and message partitioning. You can also implement the corresponding interfaces for more advanced control. A Kafka sink calls the ProducerRecord<byte[], byte[]> serialize(T element, KafkaSinkContext context, Long timestamp) method for each incoming record to generate a ProducerRecord object that represents the serialized record. Then, the Kafka sink writes the ProducerRecord object to the required topic. The ProducerRecord class provides properties that you can configure to manage behavior when the Kafka producer writes a record to the required Kafka topic. You can configure the following properties of ProducerRecord: The name of the target topic. The message key. The name of the target partition.
  Properties of the Kafka client	The bootstrap.servers property is required. This property specifies the addresses of the Kafka brokers. Separate multiple addresses with a comma (,).
  Delivery guarantee	After you enable the checkpointing feature, a Kafka sink can ensure exactly-once delivery. You can also configure the DeliveryGuarantee property of a Kafka sink to specify one of the following fault tolerance semantics: DeliveryGuarantee.NONE (default): The Kafka sink does not ensure data delivery. Data may be lost or duplicated. DeliveryGuarantee.AT_LEAST_ONCE: The Kafka sink ensures that data is not lost. However, data may be duplicated. DeliveryGuarantee.EXACTLY_ONCE: The Kafka sink ensures that data is not lost or duplicated. The Kafka transaction mechanism is used to ensure exactly-once delivery. Note For more information about the exactly-once semantics, see Usage notes of Semantic.EXACTLY_ONCE.

Usage notes of Semantic.EXACTLY_ONCE

• When you use the transaction mechanism to write data into a Kafka topic, you must configure the isolation.level option for all Kafka consumers. Valid values:

  • read_committed: Only the committed data is read.

  • read_uncommitted (default): Data that is not committed can be read.

• A Kafka sink that operates in DeliveryGuarantee.EXACTLY_ONCE mode relies on the transactions that are committed after a deployment is recovered from a checkpoint and before the checkpoint is taken. If the restart duration after deployment failure takes longer than Kafka transaction timeout duration, data loss may happen when Kafka expires an uncommitted transaction. Therefore, we recommend that you configure a transaction timeout period appropriately to your expected downtime.

  The transaction timeout that you configure for a Kafka producer cannot exceed the Kafka broker's transaction timeout period, which is set to 15 minutes by default. The Kafka sink sets the transaction.timeout.ms option in sink configuration to 1 hour by default. Therefore, you must extend the transaction timeout period for Kafka brokers before you use the DeliveryGuarantee.EXACTLY_ONCE mode.

• DeliveryGuarantee.EXACTLY_ONCE mode uses a fixed-size pool for each Kafka sink. Each producer is used per one checkpoint. If the number of concurrent checkpoints exceeds the pool size, FlinkKafkaProducer will throw an exception and fails the entire deployment. You must configure a maximum pool size and the maximum number of concurrent checkpoints accordingly.

• When a deployment starts, the Kafka sink that operates in DeliveryGuarantee.EXACTLY_ONCE execution mode tries to terminate transactions that are not committed during checkpointing. These uncommitted transactions may block consumers from reading data. However, in the event of deployment failure before the first checkpoint, the information about previous pool sizes is lost after restarting the deployment. If you want to reduce the deployment parallelism before the first checkpoint is taken, make sure that the value after you divide the original parallelism by the new parallelism is greater than or equal to the value of FlinkKafkaProducer.SAFE_SCALE_DOWN_FACTOR. Otherwise, memory leaks may occur because of untracked transactions.

• For a Kafka consumer for which the isolation.level is set to read_committed, any uncommitted transaction (neither terminated nor completed) will block all reads from the Kafka topic. Example:

  1. A user creates a transaction to write data to a topic.

  2. The user creates another transaction to write data to the topic.

  3. The user commits the second transaction.

  In this case, the Kafka consumer cannot read the data from the second transaction until the first transaction is committed or terminated. Therefore, take note of the following points:

  • Data writes into a Kafka topic experience delays, which are approximately equal to the average checkpointing interval.

  • In the event of deployment failure, the topics that are written by the deployment block Kafka consumers from reading data until the deployment is restarted or the transactions time out.

References

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

• FAQ about connectors

• Why no data output is returned after the data of an ApsaraMQ for Kafka source table is calculated by using event time-based window functions?

• How do I connect to a Kafka cluster for which security information is configured?

• How do I resolve field name conflicts?

• FAQ about workspace and namespace management and operations