YinChunGuang opened a new issue, #19313:
URL: https://github.com/apache/hudi/issues/19313
### Bug Description
## Title: Flink compaction leaves inflight instant when lock acquisition
fails during commit
### Bug Description
When using `FileSystemBasedLockProvider` with a MOR table in Flink
streaming, if lock acquisition fails during compaction commit (e.g., transient
HDFS I/O issues), the inflight instant (`.compaction.inflight`) is **not rolled
back** and remains on the timeline forever — blocking all subsequent
compactions until the job restarts.
**Production trace (Hudi 1.1.1, Flink 1.18):**
```
01:03:26 ERROR RetryHelper - Still failed to acquire lock after retried 30
times.
HoodieLockException: Unable to acquire the lock. Current lock owner
information : ← EMPTY!
at LockManager.lock(LockManager.java:77)
at TransactionManager.beginStateChange(TransactionManager.java:58)
at
HoodieFlinkTableServiceClient.completeCompaction(HoodieFlinkTableServiceClient.java:79)
at
BaseHoodieWriteClient.completeCompaction(BaseHoodieWriteClient.java:1179)
at CompactionCommitSink.doCommit(CompactionCommitSink.java:200)
01:03:26 ERROR CompactionCommitSink - Error while committing compaction
instant: 20260715005851659
... 6 hours later, after job restart ...
07:06:43 INFO CompactionUtil - Rollback the inflight compaction instant:
[==>20260715005851659__compaction__INFLIGHT] for failover
```
Note: `Current lock owner information` is **empty** — the lock file itself
was inaccessible, not held by another process.
### Root Cause
In `CompactionCommitSink.commitIfNecessary()`, the catch block only logs the
error but does **not** attempt to rollback the inflight:
```java
// CompactionCommitSink.java L165-L174
try {
doCommit(instant, events);
} catch (Throwable throwable) {
LOG.error("Error while committing compaction instant: " + instant,
throwable);
this.compactionMetrics.markCompactionRolledBack();
// ← MISSING: rollback of inflight compaction instant
} finally {
reset(instant);
}
```
Call chain when lock fails:
```
CompactOperator.doCompaction()
→ CompactionCommitSink.doCommit() // L200
→ BaseHoodieWriteClient.completeCompaction()
→ HoodieFlinkTableServiceClient.completeCompaction()
→ TransactionManager.beginStateChange()
→ LockManager.lock() → 30 retries → HoodieLockException
```
The inflight rollback currently only happens in:
- `HoodieFlinkCompactor.compact()` L246 — only at job startup / scheduled
cycle
- `CompactionUtil.rollbackEarliestCompaction()` L188 — only after timeout
expires
For a continuously running streaming job, the orphaned inflight persists
indefinitely.
### Proposed Fix
Add a rollback attempt in the catch block of `commitIfNecessary()`:
```java
try {
doCommit(instant, events);
} catch (Throwable throwable) {
LOG.error("Error while committing compaction instant: " + instant,
throwable);
this.compactionMetrics.markCompactionRolledBack();
// Attempt to rollback inflight to prevent blocking subsequent
compactions
try {
CompactionUtil.rollbackCompaction(table, instant,
writeClient.getTransactionManager());
} catch (Exception rollbackEx) {
LOG.error("Failed to rollback inflight compaction instant: " +
instant, rollbackEx);
}
} finally {
reset(instant);
}
```
**File**:
`hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java`
### Steps to Reproduce
1. Create a MOR table with `FileSystemBasedLockProvider` + Flink streaming
writes
2. Trigger a compaction
3. Make HDFS lock directory temporarily inaccessible during commit
4. `LockManager` retries 30 times → `HoodieLockException`
5. `.compaction.inflight` remains on timeline, blocking all further
compactions
6. Restart job → inflight cleaned up, compaction resumes
### Environment
**Hudi version:*1.1.1*
**Query engine:** fli
**Relevant configs:**
### Logs and Stack Trace
2026-07-15 01:03:26,005 ERROR org.apache.hudi.common.util.RetryHelper
[] - Still failed to acquire lock after retried 30 times.
org.apache.hudi.exception.HoodieLockException: Unable to acquire the lock.
Current lock owner information :
at
org.apache.hudi.client.transaction.lock.LockManager.lambda$lock$20c251e3$1(LockManager.java:83)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at org.apache.hudi.common.util.RetryHelper.start(RetryHelper.java:94)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.transaction.lock.LockManager.lock(LockManager.java:77)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.transaction.TransactionManager.beginStateChange(TransactionManager.java:58)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.HoodieFlinkTableServiceClient.completeCompaction(HoodieFlinkTableServiceClient.java:79)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.BaseHoodieWriteClient.completeCompaction(BaseHoodieWriteClient.java:1179)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.doCommit(CompactionCommitSink.java:200)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.commitIfNecessary(CompactionCommitSink.java:166)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.invoke(CompactionCommitSink.java:120)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.invoke(CompactionCommitSink.java:58)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.hudi.adapter.MaskingOutputAdapter.collect(MaskingOutputAdapter.java:60)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.adapter.MaskingOutputAdapter.collect(MaskingOutputAdapter.java:30)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.flink.table.runtime.util.StreamRecordCollector.collect(StreamRecordCollector.java:44)
~[flink-table-runtime-1.18.1.jar:1.18.1]
at
org.apache.hudi.sink.compact.CompactOperator.doCompaction(CompactOperator.java:200)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactOperator.lambda$processElement$0(CompactOperator.java:168)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:131)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
[?:1.8.0_281]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
[?:1.8.0_281]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_281]
2026-07-15 01:03:26,006 INFO
org.apache.hudi.client.transaction.TransactionManager [] - State change
ending for action instant
Option{val=[==>20260715005851659__compaction__INFLIGHT]}
2026-07-15 01:03:26,009 INFO
org.apache.hudi.client.transaction.lock.LockManager [] - Released
connection created for acquiring lock
2026-07-15 01:03:26,009 INFO
org.apache.hudi.client.transaction.TransactionManager [] - State change
ended for action instant
Option{val=[==>20260715005851659__compaction__INFLIGHT]}
2026-07-15 01:03:26,010 ERROR
org.apache.hudi.sink.compact.CompactionCommitSink [] - Error while
committing compaction instant: 20260715005851659
org.apache.hudi.exception.HoodieLockException: Unable to acquire the lock.
Current lock owner information :
at
org.apache.hudi.client.transaction.lock.LockManager.lambda$lock$20c251e3$1(LockManager.java:83)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at org.apache.hudi.common.util.RetryHelper.start(RetryHelper.java:94)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.transaction.lock.LockManager.lock(LockManager.java:77)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.transaction.TransactionManager.beginStateChange(TransactionManager.java:58)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.HoodieFlinkTableServiceClient.completeCompaction(HoodieFlinkTableServiceClient.java:79)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.client.BaseHoodieWriteClient.completeCompaction(BaseHoodieWriteClient.java:1179)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.doCommit(CompactionCommitSink.java:200)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.commitIfNecessary(CompactionCommitSink.java:166)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.invoke(CompactionCommitSink.java:120)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactionCommitSink.invoke(CompactionCommitSink.java:58)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:75)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
~[flink-dist-1.18.1.jar:1.18.1]
at
org.apache.hudi.adapter.MaskingOutputAdapter.collect(MaskingOutputAdapter.java:60)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.adapter.MaskingOutputAdapter.collect(MaskingOutputAdapter.java:30)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.flink.table.runtime.util.StreamRecordCollector.collect(StreamRecordCollector.java:44)
~[flink-table-runtime-1.18.1.jar:1.18.1]
at
org.apache.hudi.sink.compact.CompactOperator.doCompaction(CompactOperator.java:200)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.compact.CompactOperator.lambda$processElement$0(CompactOperator.java:168)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:131)
~[hudi-flink1.18-bundle-1.1.1.jar:1.1.1]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
[?:1.8.0_281]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
[?:1.8.0_281]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_281]
20
--
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]
