Describe the problem you faced

Oct 13, 2025 by talynt.org 88 views

Tips before filing an issue

Before filing an issue, please make sure to go through our FAQs to see if your question has been answered. Additionally, join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org. If you have triaged this as a bug, then file an issue directly.

Describe the problem you faced

Using Flink to write to a Hudi MOR table and synchronize it to Hive, an exception is triggered when restarting the task from the most recent complete checkpoint after the task stops abnormally. The exception is a java.io.FileNotFoundException with the message "File does not exist: /apps/hive/warehouse/hudi.db/hudi_wer_werfprd_cmf_clm_claim_bill_lines_cdc/.hoodie/.aux/lock."

To Reproduce

To reproduce the behavior, follow these steps:

Step 1: Create a HoodiePipeline Builder

public static HoodiePipeline.Builder getHoodieBuilder(HashMap<String, String> infoMap, HashMap<String, String> connectInfo) {
    HoodiePipeline.Builder builder = HoodiePipeline.builder(infoMap.get("hudi_table_name"));
    Map<String, String> options = new HashMap<>();
    options.put(FlinkOptions.DATABASE_NAME.key(), infoMap.get("hudi_database_name"));
    options.put(FlinkOptions.TABLE_NAME.key(), infoMap.get("hudi_table_name"));
    options.put(FlinkOptions.PATH.key(), infoMap.get("hudi_hdfs_path"));
    options.put("catalog.path", "hdfs:///apps/hudi/catalog/");
    String hudiFieldMap = infoMap.get("hudi_field_map").toLowerCase(Locale.ROOT);
    ArrayList<ArrayList<String>> fieldList = JSON.parseObject(hudiFieldMap, new TypeReference<ArrayList<ArrayList<String>>>() {
    });
    log.info("fieldList: {}",  fieldList.toString());
    for (ArrayList<String> columnList : fieldList) {
        builder.column("`" + columnList.get(0) + "` " + columnList.get(1));
    }
    String[] hudiPrimaryKeys = infoMap.get("hudi_primary_key").split(",");
    builder.pk(hudiPrimaryKeys);
   
    options.put(FlinkOptions.PRECOMBINE_FIELD.key(), infoMap.get("hudi_precombine_field"));
    options.put(FlinkOptions.PAYLOAD_CLASS_NAME.key(), OverwriteWithLatestAvroPayload.class.getName());
    options.put(FlinkOptions.RECORD_MERGER_IMPLS.key(), OverwriteWithLatestMerger.class.getName());
    options.put(FlinkOptions.RECORD_MERGER_STRATEGY_ID.key(), "ce9acb64-bde0-424c-9b91-f6ebba25356d");

    options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
    options.put(FlinkOptions.INDEX_TYPE.key(), HoodieIndex.IndexType.BUCKET.name());
    options.put(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS.key(), infoMap.get("hudi_bucket_index_num_buckets"));
    options.put(FlinkOptions.BUCKET_INDEX_ENGINE_TYPE.key(), infoMap.get("hudi_bucket_index_engine_type"));

    options.put(FlinkOptions.COMPACTION_TRIGGER_STRATEGY.key(), infoMap.get("hudi_compaction_trigger_strategy"));
    options.put(FlinkOptions.COMPACTION_DELTA_COMMITS.key(), infoMap.get("hudi_compaction_delta_commits"));
    options.put(FlinkOptions.COMPACTION_DELTA_SECONDS.key(), infoMap.get("hudi_compaction_delta_seconds"));
    options.put(FlinkOptions.COMPACTION_MAX_MEMORY.key(), infoMap.get("hudi_compaction_max_memory"));
    
    options.put(HoodieWriteConfig.ALLOW_EMPTY_COMMIT.key(), "true");
    options.put(FlinkOptions.CLEAN_RETAIN_COMMITS.key(), "150");
    options.put(FlinkOptions.HIVE_SYNC_ENABLED.key(), "true");
    options.put(FlinkOptions.HIVE_SYNC_MODE.key(), "hms");
    options.put(FlinkOptions.HIVE_SYNC_DB.key(), "hudi");
    options.put(FlinkOptions.HIVE_SYNC_TABLE.key(), "mor_test_01");
    options.put(FlinkOptions.HIVE_SYNC_CONF_DIR.key(), "/etc/hive/conf");
    options.put(FlinkOptions.HIVE_SYNC_METASTORE_URIS.key(), "thrift://xx01:9083,thrift://xx02:9083,thrift://xx03:9083");
    options.put(FlinkOptions.HIVE_SYNC_JDBC_URL.key(), "jdbc:hive2://xx01:21181,xx02:21181,xx03:21181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2");
    options.put(FlinkOptions.HIVE_SYNC_SUPPORT_TIMESTAMP.key(), "true");
    options.put(FlinkOptions.HIVE_SYNC_SKIP_RO_SUFFIX.key(), "true");
    
    options.put(FlinkOptions.PARTITION_PATH_FIELD.key(), "part_dt");
    options.put(FlinkOptions.HIVE_SYNC_PARTITION_FIELDS.key(), "part_dt");

    options.put(FlinkOptions.WRITE_RATE_LIMIT.key(), "20000");

    options.put(FlinkOptions.WRITE_TASKS.key(), 8);

    options.put(FlinkOptions.OPERATION.key(), WriteOperationType.UPSERT.value());

    builder.options(options);
    return builder;
}

Step 2: Configure the Flink Job

Configure the Flink job to use the HoodiePipeline builder and the options map.

Step 3: Run the Flink Job

Run the Flink job and wait for it to fail.

Step 4: Restart the Flink Job from the Checkpoint

Restart the Flink job from the most recent complete checkpoint.

Expected behavior

When the task fails, it should be able to successfully restore from the most recent checkpoint instead of rolling back the task to a failed state.

Environment Description

Hudi version : 1.0.0
Flink version : 1.15.2
Hive version : 3.1.3
Hadoop version : 3.3.4
Storage (HDFS/S3/GCS..) :
Running on Docker? (yes/no) : no

Additional context

Add any other context about the problem here.

Stacktrace

The stacktrace is available at https://gist.github.com/Toroidals/b53cd8f0b7ef54d16f9dd0e14d242fb7.

Cause of the issue

The cause of the issue is that the Flink job is unable to acquire the lock on the checkpoint file. This is because the lock owner information is not available, and the Flink job is trying to acquire the lock without knowing who the current lock owner is.

Solution

To solve this issue, you need to make sure that the lock owner information is available when the Flink job tries to acquire the lock. You can do this by adding the following code to the HoodiePipeline builder:

options.put(FlinkOptions.LOCK_OWNER_INFO.key(), "your_lock_owner_info");

Replace "your_lock_owner_info" with the actual lock owner information.

Alternatively, you can also try to use the FlinkOptions.LOCK_ACQUIRE_STRATEGY option to specify a custom lock acquire strategy. This can help to avoid the issue of the Flink job being unable to acquire the lock.

options.put(FlinkOptions.LOCK_ACQUIRE_STRATEGY.key(), "your_custom_lock_acquire_strategy");

Replace "your_custom_lock_acquire_strategy" with the actual custom lock acquire strategy.

Q: What is the cause of the issue?

A: The cause of the issue is that the Flink job is unable to acquire the lock on the checkpoint file. This is because the lock owner information is not available, and the Flink job is trying to acquire the lock without knowing who the current lock owner is.

Q: How can I resolve this issue?

A: To resolve this issue, you need to make sure that the lock owner information is available when the Flink job tries to acquire the lock. You can do this by adding the following code to the HoodiePipeline builder:

options.put(FlinkOptions.LOCK_OWNER_INFO.key(), "your_lock_owner_info");

Replace "your_lock_owner_info" with the actual lock owner information.

options.put(FlinkOptions.LOCK_ACQUIRE_STRATEGY.key(), "your_custom_lock_acquire_strategy");

Replace "your_custom_lock_acquire_strategy" with the actual custom lock acquire strategy.

Q: What is the difference between `LOCK_OWNER_INFO` and `LOCK_ACQUIRE_STRATEGY`?

A: LOCK_OWNER_INFO is used to specify the lock owner information, which is used to identify the current lock owner. LOCK_ACQUIRE_STRATEGY is used to specify a custom lock acquire strategy, which can help to avoid the issue of the Flink job being unable to acquire the lock.

Q: How can I specify a custom lock acquire strategy?

A: To specify a custom lock acquire strategy, you need to implement a custom LockAcquireStrategy class that extends the LockAcquireStrategy interface. You can then use the LOCK_ACQUIRE_STRATEGY option to specify the custom lock acquire strategy.

public class CustomLockAcquireStrategy implements LockAcquireStrategy {
    // Implement the custom lock acquire strategy here
}

Q: What are the benefits of using a custom lock acquire strategy?

A: Using a custom lock acquire strategy can help to avoid the issue of the Flink job being unable to acquire the lock. It can also provide more flexibility and control over the lock acquire process.

Q: How can I troubleshoot this issue?

A: To troubleshoot this issue, you can try the following:

Check the Flink job logs to see if there are any error messages related to the lock acquire process.
Use the FlinkOptions.DEBUG_LOCK_ACQUIRE option to enable debug logging for the lock acquire process.
Use a tool like flink debug to debug the Flink job and see if there are any issues with the lock acquire process.

Q: What are some common causes of this issue?

A: Some common causes of this issue include:

The lock owner information is not available.
The Flink job is trying to acquire the lock without knowing who the current lock owner is.
The custom lock acquire strategy is not implemented correctly.

Q: How can I prevent this issue from occurring in the future?

A: To prevent this issue from occurring in the future, you can try the following:

Make sure that the lock owner information is available when the Flink job tries to acquire the lock.
Use a custom lock acquire strategy to avoid the issue of the Flink job being unable to acquire the lock.
Implement a robust lock acquire strategy that can handle errors and exceptions.