Realtime Compute for Apache Flink:Connectors

• Kafka

  • How to retrieve JSON data from Kafka

  • Why can't Flink consume or write to Kafka with a working network connection?

  • No data output from Kafka source with an event-time window

  • Purpose of committing offsets in Kafka

  • Parsing nested JSON data with the Kafka connector

  • Connecting to a security-enabled Kafka cluster

  • Resolving field name conflicts

  • Handling unexpected latency from a Kafka source

  • Error: 'upsert-kafka' table require PRIMARY KEY

• DataHub

  • Job recovery after DataHub topic scaling

  • Can I delete a consumed DataHub topic?

  • DataHub table creation fails with NoPermissionException: dhs:ListShard

• MaxCompute

  • Understanding endPoint and tunnelEndpoint

  • Data reading for full and incremental MaxCompute sources

  • Can MaxCompute sources read data appended after job start?

  • Changing concurrency for resumed MaxCompute source jobs

  • Why does a MaxCompute source read partitions before the start position?

  • Handling incomplete new partitions in an incremental MaxCompute source

  • MaxCompute connector error: ErrorMessage=Authorization Failed [4019], You have NO privilege

  • Configuring the startPartition parameter for an incremental MaxCompute source

  • Long delay before an incremental MaxCompute source starts reading

  • Configuring the Partition parameter

  • Job with MaxCompute source stuck at startup or delayed

  • Selecting a Tunnel

  • Handling duplicate data in MaxCompute result tables

  • Runtime error in MaxCompute result table: 'Invalid partition spec'

  • Runtime error in MaxCompute result table: 'No more available blockId'

  • Difference between max_pt() and max_pt_with_done()

  • Using the SHUFFLE_HASH hint

  • Configuring the CacheReloadTimeBlackList parameter

• MySQL

  • Error: java.io.EOFException: SSL peer shut down incorrectly

• ApsaraDB RDS for MySQL

  • Type changes for bigint unsigned primary key during data sync

  • Does Flink update or insert records when writing to RDS?

  • Using GROUP BY on a unique index in an RDS table

  • Why INT UNSIGNED in MySQL requires a different type in Flink SQL

  • Error: Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1

  • Deadlock when writing to MySQL

  • Downstream schema does not change after MySQL schema update

  • Causes of the 'finish split response timeout' exception

  • Effects of schema changes during MySQL CDC full synchronization

  • Resolving job failures from unsupported schema changes during CTAS/CDAS sync

• ClickHouse

  • Data retraction and updates in ClickHouse result tables

  • Data visibility in ClickHouse result tables

• Print

  Viewing print results in the Console

• Tablestore

  JOIN on a dimension table returns no data

• ApsaraMQ for RocketMQ

  • Error: IllegalArgumentException: timeout value is negative

  • How RocketMQ detects partition changes during topic scale-out

• Hologres

  • Error: BackPressure Exceed reject Limit

  • Error: remaining connection slots are reserved for non-replication superuser connections

  • Error: no table is defined in publication

  • Relationship between checkpoint interval and data visibility for Hologres sink

  • 'permission denied for database' exception on deployment

  • 'table id ...' exception when restoring from a checkpoint

  • Data precision issues when consuming binlog data in JDBC mode

  • Errors after deleting and recreating a table with the same name

  • ClassCastException when reading from Hologres

• Log Service

  • Error: Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608

  • OOM error on TaskManager when restoring a failed job

• Paimon

  • Setting startup position for a Paimon source

  • Configuring automatic partition expiration

  • Paimon write job fails with 'Heartbeat of TaskManager timed out'

  • Paimon write job fails with 'Sink materializer must not be used with Paimon sink' error

  • Conflict errors when writing to Paimon

  • Paimon read job fails with 'File xxx not found' error

  • Paimon job fails with 'No space left on device'

  • Managing many Paimon files on OSS

  • Paimon data visibility and the checkpoint interval

  • Slow memory increase in jobs writing to Paimon

• Hudi

  • No data found in storage

  • Handling duplicate data in Hudi

  • Why a Merge On Read table only has log files

• AnalyticDB for MySQL (ADB)

  Error: multi-statement be found

• Custom connector

  • Error: No suitable driver found for

• Elasticsearch

  • Data loss or overwriting when Flink writes to Elasticsearch

Retrieve JSON data from Kafka using Flink

• To retrieve standard JSON data, see JSON Format.

• To retrieve nested JSON data, define the JSON object as a ROW type in the DDL for the source table. In the DDL for the sink table, declare the keys to retrieve. Then, use a DML statement to access the keys and extract their values. The following code provides an example:

  • Sample data

    {
        "a":"abc",
        "b":1,
        "c":{
            "e":["1","2","3","4"],
            "f":{"m":"567"}
        }
    }

  • Source table DDL

    CREATE TEMPORARY TABLE `kafka_table` (
      `a` VARCHAR,
       b int,
      `c` ROW<e ARRAY<VARCHAR>,f ROW<m VARCHAR>>  -- 'c' is a JSON object that maps to the ROW type in Flink. 'e' is a JSON array that maps to the ARRAY type.
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'xxx',
      'properties.bootstrap.servers' = 'xxx',
      'properties.group.id' = 'xxx',
      'format' = 'json',
      'scan.startup.mode' = 'xxx'
    );

  • Sink table DDL

    CREATE TEMPORARY TABLE `sink` (
     `a` VARCHAR,
      b INT,
      e VARCHAR,
      `m` varchar
    ) WITH (
      'connector' = 'print',
      'logger' = 'true'
    );

  • DML statement

    INSERT INTO `sink`
      SELECT 
      `a`,
      b,
      c.e[1], -- Flink uses 1-based indexing for arrays. This example uses index 1 to retrieve the first element. To retrieve the entire array, omit [1].
      c.f.m
    FROM `kafka_table`;

  • Results

    409  2021-04-08 10:13:11,214 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] -
    410  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka commitId: cb8625948210849f
    411  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka startTimeMs: 1617847991214
    412  2021-04-08 10:13:11,270 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Subscribed to partition(s): lb_test-0, lb_test-1, lb_test-2, lb_test-3, lb_test-4, lb_test-5
    413  2021-04-08 10:13:11,280 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 1 for partition lb_test-0
    414  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-1
    415  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-2
    416  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-3
    417  2021-04-08 10:13:11,290 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-4
    418  2021-04-08 10:13:11,291 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-5
    419  2021-04-08 10:13:11,302 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.Metadata [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Cluster ID: -1flJPwnTvuGFSuyCtU1hw
    420  2021-04-08 10:15:31,597 INFO  org.apache.flink.api.common.functions.util.PrintSinkOutputWriter [] - +I(abc,1,1,567)

Flink cannot consume or write to Kafka

• Cause

  If a forwarding mechanism, such as a proxy or port mapping, exists between Flink and Kafka, the Kafka client retrieves the internal network address of the Kafka server, not the address of the proxy. As a result, Flink can connect to the Kafka cluster but cannot consume or write data, even with an established network path.

  The connection process between the Flink Kafka connector and the Kafka server involves two steps:

  1. The Kafka client fetches metadata from the Kafka brokers. This metadata includes the network addresses of all brokers in the cluster.

  2. The Flink connector then uses these network addresses to consume or write data.

• Troubleshooting

  Follow these steps to determine if a forwarding mechanism, such as a proxy or port mapping, exists between Flink and Kafka:

  1. Use a ZooKeeper command-line tool (zkCli.sh or zookeeper-shell.sh) to log in to the ZooKeeper cluster that your Kafka cluster uses.

  2. Run the appropriate command for your cluster to retrieve the Kafka broker metadata.

     You can typically use the get /brokers/ids/0 command to retrieve Kafka broker metadata. The connection address is located in the endpoints field. For example, connect using the ZooKeeper Shell and run get /brokers/ids/0 to view the broker's registration information. Note the address configured in the endpoints field of the returned JSON:

     # bin/zookeeper-shell.sh localhost:2181
     Connecting to localhost:2181
     Welcome to ZooKeeper!
     JLine support is disabled
     WATCHER::
     WatchedEvent state:SyncConnected type:None path:null
     get /brokers/ids/0
     {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://116.62.xxx:9092"],"jmx_port":-1,"host":"116.62.xxx","timestamp":"1614840078030","port":9092,"version":4}

  3. Use a command such as ping or telnet to test connectivity from the Flink environment to the address from the endpoints field.

     A failed connection indicates that a forwarding mechanism, such as a proxy or port mapping, exists between Flink and Kafka.

• Solutions

  • Do not use a forwarding mechanism. Instead, establish a direct network path between Flink and Kafka. This allows Flink to directly connect to the endpoints listed in the Kafka metadata.

  • Contact your Kafka administrator to configure the forwarding address in the advertised.listeners property on the Kafka brokers. This ensures the Kafka client retrieves metadata that includes the correct forwarding address.

    Note

    Only Kafka versions 0.10.2.0 and later support adding a proxy address to the listeners of a Kafka broker.

  For more information about how this works, see KIP-103: Separate Internal and External traffic and Kafka client cannot connect to brokers.

If network connectivity between Flink and Kafka has been confirmed and the issue persists, check the following non-network causes:

Check 1: Start offset strategy

Verify the scan.startup.mode parameter in the WITH clause of your Kafka source table DDL. If the value is latest-offset, Flink only reads messages written after the job starts. If no new messages arrive after job startup, the job appears to consume no data.

To verify that new data is being produced after the job starts, use a Kafka consumer client to monitor the topic in real time.

Check 2: Data format mismatch

Verify that the format parameter in your Kafka source table WITH clause matches the actual encoding of messages in the Kafka topic. A format mismatch causes deserialization failures, which can result in the job silently skipping messages or producing no output.

To inspect the actual message format, use a Kafka consumer client to read raw bytes from the topic and examine the payload structure.

No data output from Kafka event time windows

• Problem

  A job produces no output when it uses a Kafka source table with an event time window.

• Cause

  An idle Kafka partition can prevent the watermark from advancing, which stops the event time window from producing output.

• Solution

  1. Ensure that all partitions receive data.

  2. To enable source idleness detection, add the following code to the Other Configurations section and save your changes. For detailed instructions, see How to configure custom job runtime parameters?.

     table.exec.source.idle-timeout: 5

     For more information about the table.exec.source.idle-timeout parameter, see Configuration.

Commit offset in Kafka

A Kafka commit offset tracks the position of processed data, ensuring consistency and reliability in stream processing by preventing data duplication or loss. When a checkpoint completes successfully, Flink commits the corresponding read offset to Kafka. If checkpointing is not enabled, or if the checkpoint interval is too long, the committed offset in Kafka can become stale, leading to data reprocessing or loss.

Parse nested JSON with Kafka connector

{"data":[{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"}]}

Connect to a secure Kafka cluster

1. In the WITH clause of your Kafka table DDL, add the security configurations for authentication and encryption. For a full list of options, see SECURITY.

   Important

   Prefix all security configuration parameters with properties.

   • This example shows how to configure a Kafka table to use the PLAIN SASL mechanism and provide a JAAS configuration.

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(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\";'
     );

   • This example shows how to use the SASL_SSL security protocol with the SCRAM-SHA-256 SASL mechanism.

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
     ) WITH (
       'connector' = 'kafka',
       ...
       'properties.security.protocol' = 'SASL_SSL',
       /* SSL configuration */
       /* Path to the truststore (CA certificate) provided by the server */
       'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
       'properties.ssl.truststore.password' = 'test1234',
       /* If client-side authentication is required, configure the path to the keystore (private key) */
       'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
       'properties.ssl.keystore.password' = 'test1234',
       /* SASL configuration */
       /* Configure the SASL mechanism as 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\";'
     );

     Note

     • If properties.sasl.mechanism is SCRAM-SHA-256, use org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule for properties.sasl.jaas.config.

     • If properties.sasl.mechanism is PLAIN, use org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule for properties.sasl.jaas.config.

2. In the Additional dependency files section for your job, upload all required files, such as certificates, public keys, and private keys.

   The platform stores uploaded files in the /flink/usrlib directory. For upload instructions, see job deployment.

   Important

   If the authentication mechanism on your Kafka broker is SASL_SSL but the client-side mechanism is SASL_PLAINTEXT, the job fails with an OutOfMemory exception during validation. To resolve this issue, ensure the client-side and server-side authentication mechanisms match.

Field naming conflicts

• Problem

  A Kafka data source serializes messages into two separate JSON strings: one for the key and one for the value. In this scenario, both the key and the value contain a field with the same name, such as the id field in the example below. Parsing this data directly into a Flink table causes a field naming conflict.

  • key

    {
       "id": 1
    }

  • value

    {
       "id": 100,
       "name": "flink"
    }

• Solution

  Use the key.fields-prefix property to avoid this issue.

  CREATE TABLE kafka_table (
    -- Define the columns for the key and value fields
    key_id INT,
    value_id INT,
    name STRING
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true',
    -- Specify the fields and data types for the key
    'key.format' = 'json',
    'key.fields' = 'id',
    'value.format' = 'json',
    'value.fields' = 'id, name',
    -- Add a prefix to fields from the key
    'key.fields-prefix' = 'key_'
  );

  Setting the key.fields-prefix property to key_ instructs the connector to add the key_ prefix to all fields from the message key. For example, the id field from the key becomes the key_id column in the Flink table. This prevents a conflict with the id field from the value, which maps to the value_id column.

  Running the query SELECT * FROM kafka_table; produces the following output:

  key_id: 1,
  value_id: 100,
  name: flink

Troubleshoot high latency from a Kafka source

• Problem

  When you read from a Kafka source table, the currentEmitEventTimeLag metric shows a value of over 50 years. For example, multiple Flink SQL jobs are in the running state, but the business latency column displays an unusually high value that exceeds 19,160 days, such as 19160d 1h 59m 28s.

• Troubleshooting

  1. First, determine if the job is a JAR job or a SQL job.

     For a JAR job, verify that your pom.xml file uses the Kafka dependency provided by Realtime Compute for Apache Flink. The open-source version of the connector does not report these metrics.

  2. Check if all partitions in the upstream Kafka topic are receiving data in real time.

  3. Check if the timestamp in the metadata of the Kafka message is 0 or null.

     The latency of a Kafka source is calculated by subtracting the message timestamp from the current time. If a message does not have a timestamp, the latency can be displayed as over 50 years. You can check the timestamp in one of the following ways:

     • For SQL jobs, you can retrieve the message timestamp by defining a metadata column. For more information, see Kafka source table.

       CREATE TEMPORARY TABLE sk_flink_src_user_praise_rt (
           `timestamp` BIGINT ,
           `timestamp` TIMESTAMP METADATA,  --Metadata timestamp.
           ts as to_timestamp (
             from_unixtime (`timestamp`, 'yyyy-MM-dd HH:mm:ss')
           ),
           watermark for ts as ts - interval '5' second
         ) WITH (
           'connector' = 'kafka',
           'topic' = '',
           'properties.bootstrap.servers' = '',
           'properties.group.id' = '',
           'format' = 'json',
           'scan.startup.mode' = 'latest-offset',
           'json.fail-on-missing-field' = 'false',
           'json.ignore-parse-errors' = 'true'
         );

     • Write a simple Java program that uses the KafkaConsumer client to read a message and inspect its timestamp.

Error: 'upsert-kafka' tables require a PRIMARY KEY

• Problem

  ) WITH (
      'connector' = 'upsert-kafka',
      'topic' = 'flow_stay_duration',
      'properties.bootstrap.servers' = 'xxx',
      'key.format' = 'avro',
      'value.format' = 'avro'
  );
      insert into sink_ad_data_device_info
  org.apache.flink.table.api.ValidationException: SQL validation failed. Unable to create a sink for writing table 'vvp.default.sink_ad_data_device_info'.
  The cause is following: 'upsert-kafka' tables require to define a PRIMARY KEY constraint. The PRIMARY KEY specifies which columns should be read from or write to the Kafka message key. The PRIMARY KEY also defines records in the 'upsert-kafka' table should update or delete on which keys.
  Table options are:
  'connector'='upsert-kafka'
  'key.format'='avro'
  'properties.bootstrap.servers'='xxx'
  'topic'='flow_stay_duration'
  'value.format'='avro'
      at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.formatException(ErrorConverter.java:123)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toErrorDetail(ErrorConverter.java:60)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toGrpcException(ErrorConverter.java:54)
      at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.validateAndGeneratePlan(FlinkSqlServiceImpl.java:979)
      at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:3283)
      at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
      at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)

• Cause

  This error occurs because the DDL lacks a primary key. When used as a sink table, the upsert-kafka connector consumes the changelog stream from the upstream logic. The connector writes INSERT and UPDATE_AFTER data to Kafka. For DELETE operations, it writes a message with a null value to indicate that the message for the corresponding key is deleted. Flink uses the primary key columns to partition the data. This guarantees that messages with the same key are ordered and that their corresponding update or delete messages land in the same partition.

• Solution

  Define a primary key in the DDL.

Recover a Flink job after topic split or scale-in

If you split or scale in a DataHub topic that a Flink job is reading from, the job will enter a failure loop and cannot recover automatically. To resolve this, restart the job.

Deleting topics with active consumers

You cannot delete or recreate a DataHub topic that has active consumers.

The endPoint and tunnelEndpoint parameters

DataHub table creation fails with NoPermissionException: dhs:ListShard

• Symptom

  When a Flink job deploys a DataHub source or sink table, it fails with an error similar to the following:

  NoPermissionException: You have no permission to perform this action. Action: dhs:ListShard

• Cause

  This error is caused by non-standard WITH parameter naming in the DataHub DDL. The DataHub connector requires credentials to be specified as accessId and accessKey. If you use the dotted forms access.id and access.key instead, the connector cannot recognize the credential fields and cannot authenticate. As a result, the connector attempts to list shards without valid credentials, and DataHub returns a NoPermissionException.

  The error message refers to a missing dhs:ListShard privilege, but the root cause is the unrecognized credential parameters — not an actual IAM permission deficiency.

• Solution

  In your DataHub DDL, rename access.id to accessId and access.key to accessKey. Drop the existing table and recreate it with the corrected WITH clause.

  Incorrect configuration:

  CREATE TABLE datahub_source (...) WITH (
    'connector'       = 'datahub',
    'endPoint'        = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'         = 'your_project',
    'topic'           = 'your_topic',
    'access.id'       = 'your-access-key-id',      -- Incorrect: dotted form not recognized
    'access.key'      = 'your-access-key-secret'   -- Incorrect: dotted form not recognized
  );

  Correct configuration:

  CREATE TABLE datahub_source (...) WITH (
    'connector'  = 'datahub',
    'endPoint'   = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'    = 'your_project',
    'topic'      = 'your_topic',
    'accessId'   = 'your-access-key-id',      -- Correct
    'accessKey'  = 'your-access-key-secret'   -- Correct
  );

  For the complete list of supported WITH parameters, see the DataHub connector documentation.

Full and incremental reads from MaxCompute sources

MaxCompute sources perform full and incremental reads through the MaxCompute Tunnel. The read throughput is limited by the MaxCompute Tunnel bandwidth.

Can MaxCompute source tables read appended data?

No. After a Flink job starts, it will not read new data appended to a source table or partition. This applies whether the source is actively reading or has finished reading. Appending data in this manner can also cause a job failover.

Concurrency changes for paused MaxCompute jobs

For MaxCompute source tables with the useNewApi option enabled (enabled by default), jobs in streaming mode support concurrency changes after they are paused and resumed. A MaxCompute source table reads matching partitions sequentially. When reading a partition, it distributes the data within that partition across the parallel operators. Changing the concurrency does not affect the data distribution for the partition being processed before the pause. The new parallelism takes effect only when the job begins to process the next partition. As a result, if a job is processing a single large partition, increasing the concurrency and resuming the job might cause only some of the MaxCompute operators to read data.

Concurrency changes are not supported for batch jobs or for jobs where the useNewApi option is set to false.

Why does MaxCompute read previous partitions when the start position is 2019-10-11 00:00:00?

Prevent incomplete data reads from new partitions

MaxCompute connector authorization error

• Error details

  During job execution, an error is displayed on the failover page or in the TaskManager.log file:

  ErrorMessage=Authorization Failed [4019], You have NO privilege'ODPS:***'

• Cause

  The user identity information specified in the MaxCompute DDL definition lacks the required permissions to access MaxCompute.

• Solution

  Authenticate using an Alibaba Cloud account, a RAM user, or a RAM role. For more information, see user authentication.

Configure the startPartition parameter

Slow start for incremental MaxCompute jobs

Set the partition parameter

Read from partitions

• Read from static partitions

  When reading from static partitions of a source table or dimension table, set the partition parameter as follows.

  Step	Description	Example
  1	For a dimension table, specify each partition as 'partition_column_name=partition_value'. The partition value must be a fixed value. For a source table, specify each partition as 'partition_column_name=partition_value'. The partition value can be a fixed value or a value that contains a wildcard (*). The wildcard can match any string, including an empty string.	To read data from the partition column dt with the value 20220901, specify dt=20220901. To read data from partitions in column dt whose values start with 202209, specify dt=202209* (applies to source tables only). To read data from partitions in column dt whose values start with 2022 and end with 01, specify dt=2022*01 (applies to source tables only). To read data from all partitions in column dt, specify dt=* (applies to source tables only).
  2	Sort the partition strings from Step 1 by partition level in ascending order, and then join them with commas (no spaces). The resulting string is the value of the partition parameter. You can specify only the first few partition levels.	A table has a single, first-level partition dt. To read data from the dt=20220901 partition, specify 'partition' = 'dt=20220901'. A table has three partition levels: a first-level partition dt, a second-level partition hh, and a third-level partition mm. To read data from dt=20220901, hh=08, and mm=10, specify 'partition' = 'dt=20220901,hh=08,mm=10'. For the same table, to read data from dt=20220901, hh=08, and any value for mm, specify 'partition' = 'dt=20220901,hh=08' or 'partition' = 'dt=20220901,hh=08,mm=*'. For the same table, to read data from dt=20220901, any value for hh, and mm=10, specify 'partition' = 'dt=20220901,hh=*,mm=10'.

  If these steps do not meet your partition filtering needs, you can add the filter conditions to the WHERE clause of your SQL statement. This allows the SQL optimizer to use partition pushdown for filtering. For example, to read partitions from a table with two partition levels (dt and hh) where dt is between '20220901' and '20220903', and hh is between '09' and '17', use a SQL statement like the following.

  CREATE TABLE maxcompute_table (
    content VARCHAR,
    dt VARCHAR,
    hh VARCHAR
  ) PARTITIONED BY (dt, hh) WITH ( 
     -- You must specify the partition columns with PARTITIONED BY to enable 
     -- partition pushdown in the SQL optimizer, which improves performance.
    'connector' = 'odps',
    ... -- Fill in required parameters such as accessId. You can omit the 'partition' parameter and let the SQL optimizer filter partitions.
  );
  SELECT content, dt, hh FROM maxcompute_table
  WHERE dt >= '20220901' AND dt <= '20220903' AND hh >= '09' AND hh <= '17'; -- Specify partition filters in the WHERE clause.

• Read the partition with the largest lexicographical order

  • To read the lexicographically largest partition from a source table or dimension table, set the partition parameter to 'max_pt()'.

  • To read the two lexicographically largest partitions from a source table or dimension table, set the partition parameter to 'max_two_pt()'.

  • To read the lexicographically largest partition that also has a corresponding .done partition from a source table or dimension table, set the partition parameter to 'max_pt_with_done()'.

  Typically, the lexicographically largest partition is the most recently created one. The max_pt_with_done() option is useful when the data in the latest partition might not be ready, and you want the dimension table to temporarily read from a slightly older but complete partition.

  When data for a partition is ready, you must also create a corresponding empty partition. Its name is the data partition name with .done appended. For example, after the data for the dt=20220901 partition is ready, create an empty partition named dt=20220901.done. When you set the partition parameter to max_pt_with_done(), the dimension table reads only from partitions that have a corresponding .done partition. Data partitions without a .done partition are temporarily ignored. For more information, see What is the difference between max_pt() and max_pt_with_done()?.

  Note

  A source table determines the partition with the largest lexicographical order only when the job starts. It stops after reading all the data and does not monitor for new partitions. If you need to continuously read new partitions, use the incremental source table mode. A dimension table checks for and reads the latest data each time it is refreshed.

Write to partitions

• Write to static partitions

  To write data to a static partition of a result table, you can set the partition parameter using the same method as for reading from static partitions.

  Important

  The partition parameter for a result table does not support wildcards (*).

• Write to dynamic partitions

  To write to dynamic partitions, where partition values are derived from the data, set the partition parameter to a comma-separated list of partition column names, sorted by partition level in ascending order. For example, if a table has three partition levels, dt, hh, and mm, specify 'partition' = 'dt,hh,mm'.

Slow job startup for MaxCompute source tables

Choose a data channel

Data duplication in MaxCompute result tables

Duplicate data in a MaxCompute result table written by a Flink job can result from the following causes:

• Check your job logic. Even if a primary key constraint is declared in the MaxCompute result table, Flink does not perform uniqueness checks when writing to external storage. Additionally, non-transactional tables in MaxCompute do not support primary key constraints. Therefore, if your Flink job logic generates duplicate data, these duplicates will be written to the MaxCompute table.

• Verify if multiple Flink jobs are writing to the same MaxCompute table concurrently. As mentioned, MaxCompute does not enforce primary key constraints. If multiple Flink jobs produce the same results, they will create duplicate records in the table.

• The Flink job fails during a checkpoint while using the Batch Tunnel. When a failure occurs during a checkpoint, data for the result table may have already been committed to the server. As a result, when the job recovers from the last successful checkpoint, it may write duplicate data for the period between the last successful checkpoint and the failure.

• A Flink job failover occurs while using the Stream Tunnel. When writing to MaxCompute with the Stream Tunnel, data is committed to the MaxCompute server between checkpoints. If the job fails over and recovers from the latest checkpoint, it may write duplicate data that was processed after the checkpoint completed but before the failover. For more information, see Choose a data channel. To prevent this type of duplication, you can switch to Batch Tunnel mode.

• A Flink job that uses the Batch Tunnel fails over or is restarted after being canceled (for example, triggered by Autopilot). In versions earlier than vvr-6.0.7-flink-1.15, the job commits data to the MaxCompute result table on shutdown. Consequently, when the Flink job stops and then recovers from the last checkpoint, it can create duplicate data for the period between the last checkpoint and the shutdown. To resolve this issue, upgrade your Flink version to vvr-6.0.7-flink-1.15 or later.

MaxCompute job fails with 'Invalid partition spec'

• Cause: This error occurs when the data written to MaxCompute contains invalid values in the partition column. Invalid values include empty strings, null values, or values that contain an equal sign (=), a comma (,), or a forward slash (/).

• Solution: Verify that the values in the partition column of your source data are valid.

'No more available blockId' error in MaxCompute jobs

• Cause: The number of blocks written to the MaxCompute result table has exceeded the limit. This is typically caused by flushing small amounts of data too frequently.

• Solution: Tune the batchSize and flushIntervalMs parameters.

Use the SHUFFLE_HASH hint

-- Create a source table and three dimension tables.
CREATE TABLE source_table (k VARCHAR, v VARCHAR) WITH ( ... );
CREATE TABLE dim_1 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_2 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_3 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
-- Specify the names of the dimension tables to distribute in the SHUFFLE_HASH hint.
SELECT /*+ SHUFFLE_HASH(dim_1), SHUFFLE_HASH(dim_3) */
k, s.v, d1.v, d2.v, d3.v
FROM source_table AS s
INNER JOIN dim_1 FOR SYSTEM_TIME AS OF PROCTIME() AS d1 ON s.k = d1.k
LEFT JOIN dim_2 FOR SYSTEM_TIME AS OF PROCTIME() AS d2 ON s.k = d2.k
LEFT JOIN dim_3 FOR SYSTEM_TIME AS OF PROCTIME() AS d3 ON s.k = d3.k;

Configure CacheReloadTimeBlackList

Error: java.io.EOFException: SSL peer shut down incorrectly

• Error details

  Caused by: java.io.EOFException: SSL peer shut down incorrectly
      at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:239) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190) ~[?:1.8.0_302]
      at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1392) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1300) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:435) ~[?:1.8.0_302]
      at com.mysql.cj.protocol.ExportControlled.performTlsHandshake(ExportControlled.java:347) ~[?:?]
      at com.mysql.cj.protocol.StandardSocketFactory.performTlsHandshake(StandardSocketFactory.java:194) ~[?:?]
      at com.mysql.cj.protocol.a.NativeSocketConnection.performTlsHandshake(NativeSocketConnection.java:101) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.negotiateSSLConnection(NativeProtocol.java:308) ~[?:?]
      at com.mysql.cj.protocol.a.NativeAuthenticationProvider.connect(NativeAuthenticationProvider.java:204) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.connect(NativeProtocol.java:1369) ~[?:?]
      at com.mysql.cj.NativeSession.connect(NativeSession.java:133) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:949) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:819) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:449) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:242) ~[?:?]
      at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:198) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider.getOrEstablishConnection(SimpleJdbcConnectionProvider.java:128) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.AbstractJdbcOutputFormat.open(AbstractJdbcOutputFormat.java:54) ~[?:?]
      ... 14 more

• Cause

  This error typically occurs when a MySQL database has the SSL protocol enabled, but the client's SSL connection is not configured correctly. For example, with MySQL driver 8.0.27 and an SSL-enabled MySQL database, this error occurs because the driver's default access method does not use SSL.

• Solution

  Append characterEncoding=utf-8&useSSL=false to the URL parameter of the MySQL dimension table. For example:

  'url'='jdbc:mysql://***.***.***.***:3306/test?characterEncoding=utf-8&useSSL=false'

Changes to MySQL bigint unsigned key type

Flink does not support the bigint unsigned data type. To prevent potential data overflow, Flink maps the bigint unsigned primary key to the decimal type. During synchronization to Hologres, the system converts the column to the text type because Hologres neither supports bigint unsigned nor allows the decimal type as a primary key.

Consider this behavior during your design and development. To keep the column as a decimal type, manually create the table in Hologres before starting the synchronization. In this table, you can either set a different column as the primary key or define no primary key. However, this approach may lead to data duplication because the original primary key is no longer enforcing uniqueness. You must then handle this issue at the application level, for example, by tolerating data duplication or by implementing deduplication logic.

Flink to RDS: Update vs. insert

If a primary key is defined in the DDL, the connector uses an INSERT INTO tablename(field1,field2, field3, ...) VALUES(value1, value2, value3, ...) ON DUPLICATE KEY UPDATE field1=value1,field2=value2, field3=value3, ...; statement. This statement inserts a new record if the primary key does not exist, or updates the existing record if it does. If no primary key is declared in the DDL, the connector inserts new records with an insert into statement.

Using a unique index with GROUP BY

• You must declare the unique index in the GROUP BY clause of your job.

• If an RDS table uses an auto-increment primary key, do not declare it as a PRIMARY KEY in the Flink job.

INT UNSIGNED mapping: MySQL to Flink SQL

The MySQL JDBC Driver maps unsigned integers to larger Java data types to preserve precision. Specifically, the driver maps MySQL INT UNSIGNED values to the Java LONG type, which Flink SQL then treats as BIGINT. Similarly, it maps MySQL BIGINT UNSIGNED values to the Java BigInteger type, which Flink SQL treats as DECIMAL(20, 0).

Error: Incorrect string value

• Error details

  Caused by: java.sql.BatchUpdateException: Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1
  at sun.reflect.GeneratedConstructorAccessor59.newInstance(Unknown Source)
  at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  at com.mysql.cj.util.Util.getInstance(Util.java:167)
  at com.mysql.cj.util.Util.getInstance(Util.java:174)
  at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:755)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:426)
  at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:796)
  at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeBatch(DruidPooledPreparedStatement.java:565)
  at com.alibaba.ververica.connectors.rds.sink.RdsOutputFormat.executeSql(RdsOutputFormat.java:488)
  ... 15 more

• Cause

  The data contains special characters or uses a character encoding that the database does not support.

• Solution

  Add characterEncoding=UTF-8 to the URL when connecting to a MySQL database via JDBC. For example: jdbc:mysql://<internal address>/<databaseName>?characterEncoding=UTF-8.

Deadlocks in MySQL (TDDL/RDS)

• Problem

  A deadlock occurs when writing data to MySQL (TDDL/RDS).

  Important

  In Realtime Compute for Apache Flink, if you use a relational database such as MySQL as a sink (via the TDDL/RDS connector), frequent writes to a table or resource can cause deadlocks.

  Assume that an INSERT operation requires acquiring two locks, (A,B), in sequence. Lock A is a range lock. There are two transactions, (T1,T2), and the table schema is (id(auto-incrementing primary key),nid(unique key)). T1 contains two statements, insert(null,2),(null,1), while T2 contains one statement, insert(null,2).

  1. At time t, T1 executes its first INSERT statement. T1 now holds both locks (A,B).

  2. At time t+1, T2 starts an insert operation and needs to wait for lock A to lock the range (-inf,2]. At this time, lock A is held by T1 and has locked the range (-inf,2]. Because the ranges have an inclusion relationship, T2 depends on T1 to release A.

  3. At time t+2, T1 executes its second INSERT statement, which requires lock A on the range (-inf,1]. Because this range is a subset of (-inf,2], T1 must queue and wait for T2 to release the lock. As a result, T1 becomes dependent on T2 to release lock A.

  A deadlock occurs because T1 and T2 are now waiting for each other to release their respective locks.

• RDS/TDDL and Tablestore use different locking mechanisms.

  • RDS/TDDL: InnoDB's row lock is applied to the index, not to an individual record. As a result, even when accessing different rows, a lock conflict can occur if the rows share the same index key. This can prevent updates across an entire data range.

  • Tablestore: Uses a single row lock, which does not affect updates to other data.

• Solutions for deadlocks

  For high QPS/TPS or high-concurrency write scenarios, use Tablestore as the result table to prevent deadlocks. In general, using TDDL or RDS as the result table for a Flink job is not recommended.

  If you must use a relational database like MySQL as a sink node, consider the following recommendations:

  • Ensure that no other workloads are reading from or writing to the same tables.

  • If the job data volume is small, try writing data in a single-threaded manner. However, in high QPS/TPS and high-concurrency scenarios, this approach reduces write performance.

  • Avoid using a unique key if possible, as writing to tables with a unique key can cause deadlocks. If your business requirements mandate a unique key, define it by ordering the columns from most to least selective. This significantly reduces the probability of a deadlock. For example, you can place a column with an MD5 hash before the day_time(20171010) column to ensure the unique key is defined with the most selective columns first.

  • Use database sharding and table splitting based on your workload characteristics to distribute writes across multiple tables. For implementation details, contact your database administrator.

Downstream table structure update failure

Table structure synchronization does not track DDL statements. Instead, it detects a schema change by comparing consecutive records. If a DDL change occurs without any subsequent upstream data changes, the downstream table structure will not update. For details, see synchronization policy for table structure changes.

Finish split response timeout error

This error occurs when high CPU utilization on a task prevents it from responding to RPC requests from the coordinator. To resolve this, increase the CPU resources for the TaskManager on the resource configuration page.

Impact of schema changes during full load

A schema change during the full load phase can cause the job to fail or prevent the schema change from being synchronized. To resolve this, stop the job, delete the downstream table, and then restart the job without state.

Unsupported schema changes during CTAS/CDAS synchronization

Resynchronize the data for the table. To do this, stop the job, drop the downstream table, and then restart the synchronization job with a stateless start. Avoid making such incompatible changes, as the job will fail again upon restart. For details about supported schema changes, see CREATE TABLE AS (CTAS) statement.

Retraction updates in ClickHouse

Retraction updates are supported for a ClickHouse result table if you specify a primary key in the DDL for the Flink result table and set the ignoreDelete parameter to false. However, this causes a significant drop in performance.

ClickHouse is a columnar database management system designed for online analytical processing (OLAP), and its support for UPDATE and DELETE operations is limited. If you specify a primary key in the Flink DDL, the connector attempts to use ALTER TABLE UPDATE and ALTER TABLE DELETE to update and delete data. These operations are highly inefficient.

Data visibility in ClickHouse

• For a ClickHouse result table with exactly-once semantics disabled (the default), data becomes visible once the buffer is flushed. The system automatically flushes this buffer when the number of records reaches the batchSize value or the time since the last write exceeds flushIntervalMs. You do not need to wait for a checkpoint to complete.

• For a ClickHouse result table with exactly-once semantics enabled, data becomes visible only after the corresponding checkpoint successfully completes.

View print results

Dimension table join returns no data

Ensure that the schema—including data types and column names—in the DDL statement is consistent with that of the physical table.

max_pt() and max_pt_with_done()

Paimon write job error: "Heartbeat of TaskManager timed out"

The most likely cause of this error is insufficient heap memory on the TaskManager. Paimon primarily uses heap memory in the following ways:

• Each parallel instance of the writer operator for a Paimon primary key table has a memory buffer for sorting. The size of this buffer is controlled by the write-buffer-size table property, which defaults to 256 MB.

• Paimon uses the ORC file format by default, which requires an additional memory buffer to convert in-memory data to a columnar format in batches. The size of this buffer is controlled by the orc.write.batch-size table property, which defaults to 1024, meaning the buffer holds 1024 rows of data.

• Each modified bucket has a dedicated writer object to write its data.

Based on these usage patterns, here are potential causes of insufficient heap memory and their solutions:

• The value of write-buffer-size is too large.

  Try reducing this parameter. However, a buffer that is too small can lead to frequent disk writes and trigger small file compactions more often, which impacts write performance.

• A single data record is too large.

  For example, if a record contains a 4 MB JSON field, the ORC buffer can grow to 4 MB × 1024 = 4 GB, consuming significant heap memory. You have two solutions:

  • Reduce the value of orc.write.batch-size.

  • If you do not need to perform ad-hoc queries (OLAP) on the Paimon result table and only require batch or streaming consumption, you can set the 'file.format' = 'avro' and 'metadata.stats-mode' = 'none' table properties during table creation. This switches the table to the Avro format and disables statistics collection.

    Note

    These parameters can only be set during table creation. They cannot be modified with an ALTER TABLE statement or a SQL hint after the table is created.

• Writing to too many partitions simultaneously or having too many buckets per partition creates an excessive number of writer objects.

  Review your partition column configuration to ensure it is appropriate. Ensure that incorrect SQL is not causing unexpected data to be written to the partition column. Also, verify that the number of buckets is reasonable. As a best practice, the total data size per bucket should be around 2 GB and should not exceed 5 GB. For details on adjusting the number of buckets, see Adjust the number of buckets for a fixed-bucket table.

Error: "Sink materializer must not be used with Paimon sink"

The sink materializer operator handles out-of-order data from cascading joins in streaming jobs. However, in jobs that write to a Paimon table, this operator introduces overhead and can cause incorrect results when aggregation is used. Do not use the sink materializer operator with a Paimon sink.

You can disable the sink materializer operator by setting the table.exec.sink.upsert-materialize parameter to false using a SET statement or as a runtime parameter. If you also need to handle out-of-order data, see out-of-order data handling.

Paimon: File deletion conflicts detected or LSM conflicts detected error

This error can occur for the following reasons:

• Multiple jobs are writing to the same partition of the same Paimon table. In this case, Paimon resolves the conflict through failover and restart. This is expected behavior, and no action is required if the error does not recur.

• The job is restored from an outdated state, which causes the error to recur. To resolve this, restore the job from its latest state or start it without state.

• Paimon does not support separate writes from multiple INSERT statements within a single job. Instead, use a UNION ALL statement to write multiple data streams to the Paimon table.

• The concurrency of the Global Committer node or the Compaction Coordinator node (when writing to an Append Scalable table) is greater than 1. The concurrency for these nodes must be 1 to ensure data consistency.

"File xxx not found" error in Paimon consumption job

Consumption of Paimon tables relies on snapshot files. If the snapshot retention period is too short or the consumption job is inefficient, snapshot files can expire and be deleted before the job finishes. This causes the consumption job to fail.

To resolve this issue, you can adjust the snapshot file retention period, specify a consumer ID, or optimize the consumption job. To check which snapshot files are available and their creation timestamps, see the Snapshots system table.

Paimon job error: No space left on device

• An excessive number of cache files can cause this error if your job runs a Paimon query, such as using a Paimon table as a dimension table or setting changelog-producer='lookup'. To prevent this, use SQL Hints to set the following parameters to limit the query cache's maximum disk space and retention time.

  • lookup.cache-max-disk-size: The maximum local disk space that the query cache can use. Recommended values include 256 MB, 512 MB, and 1 GB.

  • lookup.cache-file-retention: The retention period for query cache files. Recommended values include 30 min, 15 min, or a shorter interval.

• For jobs that write to a Paimon table, use SQL Hints to set the following parameters. These settings limit the size of local temporary files during the write process, preventing disk space shortages.

  • write-buffer-spillable: Controls whether the write buffer can spill to disk. Setting this to false completely prevents the buffer from using any disk space.

  • write-buffer-spill.max-disk-size: The maximum disk space that the write buffer can use when it spills. Recommended values include 256 MB, 512 MB, and 1 GB.

Manage Paimon files on OSS

• Paimon retains historical data files to access previous versions of a table. You can adjust the retention policy for these files to manage storage. For detailed instructions, see Clean up expired data.

• An inappropriate partition column configuration or an excessive number of buckets can also cause this issue. As a best practice, aim for a data size of approximately 2 GB per bucket, with a maximum of 5 GB. For more information, see Bucketing.

• By default, data files are saved in the ORC format. To reduce the total data file size, you can use the ZSTD compression format by setting the table parameter 'file.compression' = 'zstd' when you create the table.

  Note

  This parameter can only be set during table creation and cannot be modified later with an ALTER TABLE statement or a SQL hint.

Data visibility depends on the checkpoint interval

Yes. Paimon relies on checkpoints to guarantee exactly-once semantics. Data is committed and becomes visible downstream only upon checkpoint completion. Before this commit, data in the local buffer is flushed to the remote file system, but it is not yet readable.

Slow memory increase in long-running Paimon jobs

• An increase in memory usage is expected if the job's rps is also slowly increasing.

• If you use a Paimon filesystem catalog to read from or write to OSS, make sure to configure the fs.oss.endpoint, fs.oss.accessKeyId, and fs.oss.accessKeySecret catalog parameters. Otherwise, the Flink job may experience a slow memory leak, which is a known issue in the community.

IllegalArgumentException: timeout value is negative

• Error details

  2021-02-24 15:14:58
  java.lang.RuntimeException: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.run(ParallelReader.java:166)
      at com.alibaba.ververica.connectors.common.source.AbstractParallelSourceBase.run(AbstractParallelSourceBase.java:205)
      at com.alibaba.ververica.connectors.metaq.source.MetaQRowDataSource.run(MetaQRowDataSource.java:84)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
  Caused by: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.runImpl(ParallelReader.java:244)

• Cause of the error

  When no new MQ messages are consumed, the MetaQSource thread sleeps for an interval defined by the pullIntervalMs parameter, which defaults to -1. The job then fails with an IllegalArgumentException because a sleep duration cannot be negative.

• Solution

  Set the pullIntervalMs parameter to a non-negative value.

Partition change discovery

• For versions of Realtime Compute for Apache Flink earlier than 6.0.2, the source operator fetches the number of partitions every 5 to 10 minutes. A failover is triggered if the number of partitions differs for three consecutive checks. As a result, the source initiates a failover within 10 to 30 minutes. After the job restarts, it reads from the updated set of partitions.

• For Realtime Compute for Apache Flink versions 6.0.2 and later, the source operator fetches the number of partitions every 5 minutes by default. When new partitions are discovered, they are assigned directly to the source operator on the TaskManager, which then begins reading data. No job failover is required, allowing the source to detect partition changes within 1 to 5 minutes.

Error: Backpressure exceeds reject limit

• Error details

  26      at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
  27      at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:911)
  28      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:879)
  29      ... 13 more
  30  Caused by: java.lang.RuntimeException: Rpc Exception failed errorCount=6 with RpcException: request niagara.table.proto.UpsertRecordBatchRequest@b53e2d74 failed on final try 4, maxAttempts=4, sn=11.117.xxx, errorCode=11, msg=BackPresure Exceed Reject Limit [method:UpsertRecordBatch,transaction_id:xxx,table_id:xxx,table_version:128, actor_id:74538 xxx,worker_address:11.117.xxx]
  31          at com.alibaba.ververica.connectors.hologres.sink.HologresOutputFormat.sync(HologresOutputFormat.java:264)
  32          at com.alibaba.ververica.connectors.common.sink.OutputFormatSinkFunction.snapshotState(OutputFormatSinkFunction.java:91)
  33          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:128)
  34          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:101)
  35          at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
  36          at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:186)
  37          ... 23 more

• Cause

  The write pressure on the Hologres instance is too high.

• Solution

  Contact Hologres technical support with your instance information to request an upgrade.

Error: remaining connection slots are reserved for non-replication superuser connections

• Error details

  Caused by: com.alibaba.hologres.client.exception.HoloClientWithDetailsException: failed records 1, first:Record{schema=org.postgresql.model.TableSchema@188365, values=[f06b41455c694d24a18d0552b8b0****, com.chot.tpfymnq.meta, 2022-04-02 19:46:40.0, 28, 1, null], bitSet={0, 1, 2, 3, 4}},first err:[106]FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:406) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more
  Caused by: com.alibaba.hologres.org.postgresql.util.PSQLException: FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2553) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.readStartupMessages(QueryExecutorImpl.java:2665) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.<init>(QueryExecutorImpl.java:147) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.ConnectionFactoryImpl.openConnectionImpl(ConnectionFactoryImpl.java:273) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.ConnectionFactory.openConnection(ConnectionFactory.java:51) ~[?:?]
      at com.alibaba.hologres.org.postgresql.jdbc.PgConnection.<init>(PgConnection.java:240) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.makeConnection(Driver.java:478) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.connect(Driver.java:277) ~[?:?]
      at java.sql.DriverManager.getConnection(DriverManager.java:674) ~[?:1.8.0_302]
      at java.sql.DriverManager.getConnection(DriverManager.java:217) ~[?:1.8.0_302]
      at com.alibaba.hologres.client.impl.ConnectionHolder.buildConnection(ConnectionHolder.java:122) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:195) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:184) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.doHandlePutAction(Worker.java:460) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:389) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more

• Cause

  The connection limit for the Hologres instance has been exceeded.

• Solution

  • Check the app_name of connections to each Frontend (FE) to count the Hologres client connections from the flink-connector.

  • Check for other jobs connected to Hologres.

  • Release connections. For more information, see connection management.

No table is defined in publication

• Error details

  Dropping and recreating a table with the same name may cause a job to report no table is defined in publication.

• Cause

  Dropping a table does not delete its associated publication.

• Solution

  1. In Hologres, execute the select * from pg_publication where pubname not in (select pubname from pg_publication_tables); command to query for publication information that was not also cleaned up when a table was deleted.

  2. Execute the drop publication xx; statement to drop the remaining publication.

  3. Restart the job.

Checkpoint interval and data visibility

The checkpoint interval of the Flink Hologres sink connector does not directly control data visibility in Hologres. Its primary role is to define the SLA for failure recovery.

The Hologres connector does not support transactions. It periodically flushes the in-memory buffer to the database. A checkpoint ensures that all data is flushed by the time it completes, but the connector does not wait for the entire interval to pass before flushing. The connector triggers a flush earlier if certain buffer conditions are met (for more information, see Hologres, Hologres, and Hologres). Because a data warehouse typically does not require transactional consistency, the connector flushes data asynchronously in the background. It then performs a final, forced flush during each checkpoint to prepare for failure recovery.

Job deployment throws permission denied for database error

• Cause

  Starting with Realtime Compute for Apache Flink VVR 8.0.4, the connector enforces JDBC mode to consume the binary log from Hologres instances V2.0 or later. For these instances, non-superuser accounts require special permissions to consume the binary log in JDBC mode.

• Solution

  Grant the non-superuser account permissions to consume the binary log in JDBC mode.

  user_name refers to an Alibaba Cloud account ID or a RAM user. For more information, see account overview.

  -- For the expert permission model, grant the CREATE permission and the replication role to the user.
  GRANT CREATE ON DATABASE <db_name> TO <user_name>;
  alter role <user_name> replication;
  -- If the database uses the simple permission model (SPM), you cannot run GRANT statements. 
  -- Instead, use spm_grant to grant the user the Admin role for the database. You can also grant permissions directly in HoloWeb.
  call spm_grant('<db_name>_admin', '<user_name>');
  alter role <user_name> replication;

Job recovery failure: table id parsed from checkpoint is different from the current table id

• Cause

  This exception occurs in Realtime Compute for Apache Flink versions VVR 8.0.5 to VVR 8.0.8. When a job using a Hologres binlog source table recovers from a checkpoint, the engine enforces a strict check on the table id. If the current table id of the Hologres table does not match the one stored in the checkpoint, the recovery fails. This indicates that the source table was truncated or recreated while the job was running.

• Solution

  Upgrade to VVR 8.0.9 or a later version and restart the job. VVR 8.0.9 removes the strict check on the table id to accommodate complex business scenarios. However, avoid recreating a binlog source table. When a table is recreated, its entire binlog history is cleared. If Flink then uses the consumer offset from the old table to read data from the new table, it can cause data inconsistencies.

Unexpected Binlog data precision in JDBC mode

• Cause

  In Realtime Compute for Apache Flink 8.0.10 and earlier, unexpected data precision occurs if the precision of the DECIMAL type declared in a Flink DDL for a Binlog source table does not match the precision in Hologres.

• Solution

  This issue is fixed in Realtime Compute for Apache Flink 8.0.11. However, ensure that the precision of the DECIMAL type is consistent between Flink and Hologres to prevent precision loss.

Deleting and recreating a table with the same name may cause jobs to report no table is defined in publication or The table xxx has no slot named xxx exceptions

• Cause

  This occurs because dropping the table does not remove its associated publication.

• Solution

  Solution 1: In Hologres, run the select * from pg_publication where pubname not in (select pubname from pg_publication_tables); statement to find publications left over from dropped tables. Then, run the drop publication xx; statement to remove them. Finally, restart the job.

  Solution 2: Use VVR 8.0.5 or later. The connector automatically handles the cleanup.

ClassCastException when reading from Hologres

• Error details

  The error message is similar to the following:

  java.lang.ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base of loader 'bootstrap')

• Cause

  This error occurs when a field's type in the Flink DDL mismatches the corresponding field's type in the Hologres physical table. For example, a field is defined as BIGINT in the Flink DDL, but the corresponding field in the Hologres table is INTEGER. Since type checking is skipped for NULL values, the job throws the exception only when it reads actual data.

• Solution

  See the Hologres Data type summary documentation to ensure the field types in your Flink DDL match those in the Hologres physical table.

LogSizeTooLargeException

• Error details

  Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.ensureValidLogSize(LogAccumulator.java:249)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.doAppend(LogAccumulator.java:103)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.append(LogAccumulator.java:84)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:385)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:308)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:211)
  at com.alibaba.ververica.connectors.sls.sink.SLSOutputFormat.writeRecord(SLSOutputFo
  rmat.java:100)

• Cause

  This error occurs because a single-line log sent to Log Service exceeds the 8 MB size limit.

• Solution

  To skip the oversized log entry, change the startup position. For details, see job startup.

TaskManager OOM: Java heap space error during recovery

• Cause

  This issue is typically caused by an oversized SLS message body. The SLS connector requests data in batches. The number of LogGroups per batch is controlled by the batchGetSize parameter, which defaults to 100. This means each request can retrieve up to 100 LogGroups. During normal operation, the Flink program consumes data promptly and rarely retrieves a full batch of 100 LogGroups. However, during a failover, a large amount of unconsumed data can accumulate. If the memory required for a full batch of 100 LogGroups exceeds the JVM's available memory, the TaskManager experiences an OOM.

• Solution

  Reduce the value of the batchGetSize parameter.

Set consumption offset for a Paimon source table

To set the consumption offset for a Paimon source table, use the scan.mode parameter. The following table describes the available values and their behavior.

Configure automatic partition expiration

Paimon tables can automatically delete partitions whose time to live exceeds the specified partition expiration time. This feature helps reduce storage costs. The process is as follows:

• Time to live: The current system time minus the timestamp derived from the partition value. The partition value is converted to a timestamp as follows:

  1. Convert a partition value to a time string using the format string specified by the partition.timestamp-pattern parameter.

     In this format string, a partition column is represented by a dollar sign ($) followed by the column name. For example, if the partition columns are year, month, day, and hour, the format string $year-$month-$day $hour:00:00 converts the partition year=2023,month=04,day=21,hour=17 to the string 2023-04-21 17:00:00.

  2. Convert the time string to a timestamp using the format string specified by the partition.timestamp-formatter parameter.

     If this parameter is not set, the system defaults to the yyyy-MM-dd HH:mm:ss and yyyy-MM-dd formats. You can use any format string compatible with Java's DateTimeFormatter.

• Partition expiration time: The value you configure for the partition.expiration-time parameter.

Troubleshooting missing data in storage

• Data may not be visible in storage immediately. A Flink writer flushes data to disk under the following conditions:

  • Data buffered in a bucket reaches a certain size (default: 64 MB).

  • The total buffer size reaches a threshold (default: 1 GB).

  • A checkpoint is triggered, which flushes all in-memory data.

• If you are using stream writing, ensure that checkpointing is enabled.

Handling duplicate data in Hudi

• For COW writes, enable the write.insert.drop.duplicates parameter.

  By default, a COW write does not deduplicate data in the first file of each bucket and only applies deduplication to incremental data. To perform global deduplication, you must enable this parameter. For a MOR write, no extra parameters are needed. Defining a primary key enables global deduplication by default.

  Note

  As of Hudi version 0.10.0, this property is renamed to write.precombine and is set to true by default.

• To perform deduplication across multiple partitions, set the index.global.enabled parameter to true.

  Note

  • As of Hudi version 0.10.0, this property is set to true by default.

  • When index.type=bucket, setting the index.global.enabled parameter to true is ineffective because Bucket indexes do not support cross-partition changes. Therefore, even if the global index is enabled, the deduplication feature cannot work across multiple partitions.

• For long-window updates, such as modifying data from a month ago, increase the index.state.ttl parameter, which is measured in days.

  The index is Hudi's core data structure for identifying duplicate data. The index.state.ttl parameter controls how long the index state persists. The previous default value was 1.5 days. A value of 0 or less indicates that the index state is retained permanently.

  Note

  As of Hudi version 0.10.0, this property defaults to 0.

Merge On Read has only log files

• Cause: Hudi creates Parquet files only after compaction; otherwise, it only creates log files. By default, Merge On Read tables use asynchronous compaction, which triggers a compaction job after every five commits.

• Solution: Adjust the compaction.delta_commits parameter to trigger compaction jobs sooner.

Error: "multi-statement be found"

• Problem

  A Flink job that writes data to an AnalyticDB for MySQL (ADB) instance fails and restarts. The logs show an error similar to the following: Caused by: java.sql.SQLSyntaxErrorException: [13000, 2024101216171419216823505703151806929] multi-statement be found.

  at java.util.TimerThread.run(Timer.java:505)
  Caused by: java.sql.BatchUpdateException: [13000, 2024101216400819216823505703151079281] multi-statement be found.
  	at sun.reflect.GeneratedConstructorAccessor115.newInstance(Unknown Source)
  	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  	at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  	at com.mysql.cj.util.Util.getInstance(Util.java:167)
  	at com.mysql.cj.util.Util.getInstance(Util.java:174)
  	at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executePreparedBatchAsMultiStatement(ClientPreparedStatement.java:584)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:431)
  	at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:795)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement...)
  	at com.ververica.connectors.mysql.table.sink.MySqlOutputFormat.executeSql(MySqlOutputFormat.java:567)
  	... 6 more

• Cause

  This error indicates a compatibility issue between version 8.x of the MySQL JDBC driver and an AnalyticDB for MySQL (ADB) database with ALLOW_MULTI_QUERIES=true enabled.

• Solution

  1. Contact technical support to obtain a custom ADB 3.0 connector that uses version 5.1.46 of the MySQL JDBC driver. Apply this connector to your Flink task. See Manage custom connectors for instructions on using a custom connector.

  2. Set the allowMultiQueries=true parameter in the URI of the ADB table, for example, jdbc:mysql://xxxxx.ads.aliyuncs.com:3306/xxx?allowMultiQueries=true.

Error: No suitable driver found

• Cause

  The custom connector cannot find the required driver.

• Solution

  • Load the driver in the factory class by calling Class.forName.

  • Add the driver as an additional dependency and set the kubernetes.application-mode.classpath.include-user-jar parameter to true. See How do I configure custom job run parameters? for instructions.

Data loss or overwriting when Flink writes to Elasticsearch

• Cause 1: Conflict between doc_as_upsert and an Elasticsearch Ingest Pipeline

  When both of the following are configured simultaneously, Elasticsearch may process partial updates and pipeline transformations in an incompatible order, causing documents to be unexpectedly overwritten or lost:

  • sink.bulk-flush.update.doc_as_upsert = 'true' in the Flink DDL WITH clause

  • An Elasticsearch Ingest Pipeline assigned to the target index

  In this configuration, Elasticsearch applies the Ingest Pipeline before processing the partial update. Depending on the pipeline logic and Elasticsearch version, this interaction can cause document fields to be overwritten with incorrect values or documents to be silently dropped.

• Cause 2: Multiple Flink sink tasks writing to the same index

  If two or more Flink jobs — or multiple parallel sink instances with non-overlapping key ranges — write to the same Elasticsearch index without coordinated key handling, documents can overwrite each other. This results in data loss for whichever sink writes last to a given document ID.

• Solutions

  • For the doc_as_upsert and Ingest Pipeline conflict: Remove the sink.bulk-flush.update.doc_as_upsert = 'true' parameter from the Flink DDL, and remove or reassign the Elasticsearch Ingest Pipeline from the target index. Move any data transformation logic that was handled by the pipeline into the Flink job itself — for example, using a ProcessFunction or computed columns before the sink operator.

  • For multiple sink tasks overwriting each other: Ensure each Flink job writes to a separate Elasticsearch index, or consolidate writes into a single Flink job that manages document keys deterministically.