Help Center/ MapReduce Service/ Component Operation Guide (LTS)/ Using Flink/ Enterprise-Class Enhancements of Flink/ Offset Restoration from a Group and State Restoration from a Checkpoint in the Flink SQL Kafka Connector
Updated on 2026-06-27 GMT+08:00
View PDF
Share

Offset Restoration from a Group and State Restoration from a Checkpoint in the Flink SQL Kafka Connector

This section applies only to MRS 3.6.0-LTS or later.

Scenarios

When you configure a Flink job, Kafka job recovery from a checkpoint supports restoring offsets from the consumer group and restoring state from the checkpoint.

Notes and Constraints

  • Ensure that the Kafka topic names are consistent before and after checkpoint restoration.
  • Make sure the Kafka group offset matches the checkpoint record. Otherwise, data may be missing.

How to Use

In the created Kafka Connector Source stream table, add the restore.mode setting and set its value to group-offsets.

Example SQL statements:

CREATE TABLE KafkaSource (
  `user_id` VARCHAR,
  `user_name` VARCHAR,
  `age` INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'test_source',
  'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv',
  'properties.sasl.kerberos.service.name' = 'kafka',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.kerberos.domain.name' = 'hadoop.System domain name',
  'restore.mode' = 'group-offsets'
);
CREATE TABLE KafkaSink(`user_id` VARCHAR, `value` BIGINT) WITH ('connector' = 'print');
Insert into
  KafkaSink
select
  user_id,
  sum(age) as `value`
from
  KafkaSource
group by
  user_id;