Joy-2000 opened a new issue, #19587: URL: https://github.com/apache/hudi/issues/19587
### Describe the problem you faced ## Background In our production environment, we mainly run the following types of Flink streaming write workloads: - COW tables with append writes - MOR tables with Bucket Index and upsert writes A single Hudi table may be concurrently accessed by one or more write, table-service, and backfill jobs, including: - Flink streaming jobs writing to the latest date partition, potentially with asyn table services enabled - Spark jobs scheduling and/or executing clustering, compaction, log compaction, or cleaning - Multiple Spark jobs backfilling historical date partitions in parallel Hudi currently uses its transaction mechanism—NBCC/OCC combined with locking—to support these multi-writer scenarios. Among other things, this mechanism ensures that: - Completion times on the active timeline increase monotonically - Metadata Table updates are serialized ## Problem The current transaction boundary appears to cover several relatively expensive operations. For Flink streaming writes in particular, table-service scheduling, metadata commits, and instant-time generation share the same lock. When the underlying filesystem, such as HDFS, becomes busy or unstable, these operations may hold the lock for an extended period. This can significantly affect the stability of streaming jobs. For example, lock contention may block the sink from obtaining a new instant, resulting in prolonged backpressure, data accumulation, and delayed data availability. We would like to discuss whether the scope between `txnManager.beginStateChange` and `txnManager.endStateChange` can be reduced under NBCC. The following operations are possible candidates for moving partially or entirely outside the transaction boundary. ## 1. Table-service scheduling `scheduleTableServiceInternal` currently performs the entire scheduling process for clustering, compaction, and log compaction inside a transaction. The scheduling process may involve several expensive operations, including: - Loading the timeline - Building the filesystem view - Building the table-service plan The execution time of these operations can vary significantly, especially when the filesystem is under load. ### Proposal Keep only the operations that require serialization inside the lock, such as `createTable` and `createNewInstantTime`, and move plan building and other expensive scheduling operations outside the lock. ### Potential risk Without locking the entire scheduling process, multiple table-service jobs may schedule overlapping work concurrently. If this is not detected or rejected later, it could result in duplicate processing or data correctness issues (In our production environment, we use a separate scheduling lock to prevent table services from being scheduled concurrently). ### Expected benefit Reduce the duration for which the transaction lock is held during table-service scheduling. ## 2. Request-time generation in `startCommit` The generation of both request times and completion times currently requires acquiring the lock. Under the current timeline model, locking completion-time generation is necessary to preserve monotonically increasing completion times. However, it may be worth discussing whether the locking requirement for request-time generation can be relaxed. Compaction request times are an exception because they participate in FileSlice boundaries and therefore still require strict serialization. ### Proposal Investigate whether request-time generation for regular writes and other applicable actions can be moved outside the transaction lock, while retaining locking for: - Completion-time generation - Compaction request-time generation - Any action whose request time affects FileSlice boundaries or timeline ordering semantics ### Potential risk Concurrent writers might generate the same request time. Atomic file creation by the filesystem currently ensures that only one writer can successfully create a requested-instant file with a given name. However, the losing writer would still need to detect the collision, generate a new request time, and retry. Therefore, moving request-time generation outside the lock may require an explicit collision-detection and retry mechanism. ### Expected benefit When a Flink sink obtains its first or next instant, it would no longer be blocked solely because another operation is holding the transaction lock. This could make data consumption and checkpoint processing smoother. ## 3. `finalizeWrite` during commit `commitStats`, `completeClustering`, `completeCompaction`, and `completeLogCompaction` perform several operations while completing a write or table service. These operations include, but are not limited to: - `resolveWriteConflict` - `finalizeWrite` - `writeTableMetadata` - `saveAsComplete` `finalizeWrite` removes speculative or invalid files based on the markers for the current instant. When a write touches a large number of partitions or files, `DirectWriteMarkers` may issue a large number of filesystem `list` operations. As a result, `finalizeWrite` may hold the transaction lock for a long time when the filesystem is busy. ### Proposal Move `finalizeWrite` outside the transaction boundary, while keeping conflict resolution, Metadata Table updates, and timeline completion inside the transaction where serialization is required. ### Potential risk We currently do not see an obvious correctness dependency that requires `finalizeWrite` itself to run under the transaction lock. However, this needs to be verified for all marker implementations and failure-recovery paths. In particular, we should confirm that moving it outside the transaction does not introduce races with rollback, cleaning, or another writer operating on the same files. ### Expected benefit Reduce the lock duration caused by filesystem operations during `finalizeWrite`, especially for large writes or when HDFS is under heavy load. ## Questions for discussion 1. Which operations inside the current NBCC transaction boundary strictly require serialization? 2. Which action types can request time generation safely occur outside the lock? 3. Are there other expensive filesystem or timeline operations that can be moved outside the transaction boundary? 4. Would it make sense to introduce metrics for transaction lock wait time and lock hold time so that lock contention can be observed more directly? Our understanding and the ideas proposed above may be incomplete or incorrect. We would greatly appreciate any corrections, feedback, or suggestions from the community. ### To Reproduce none ### Expected behavior The overall goal is to preserve the existing NBCC correctness guarantees while reducing lock contention and improving the stability of Flink streaming writes in multi-writer environments. ### Environment Description Hudi：1.1.1 Flink：1.20 Spark：3.4 ### Additional context _No response_ ### Stacktrace ```shell ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
