[Improve][Zeta] Reduce checkpoint state-store latency variance by removing redundant WAL syncs - #12081
Conversation
…edundant WAL sync (apache#12058)
nzw921rx
added
the
performance
|
Hello, may I ask how to resolve the fluctuation of the Error confidence interval? 
|
| Barrier 传递、任务快照、ACK 等待、Fixture 生成、持久性校验和清理均不计入测量。每次 | ||
| invocation 固定执行 100 个逻辑操作，并按单个操作归一化为 `us/op`；数值越低越好。 | ||
|
|
||
| 上述两个隔离方法都走 write-through IMap MapStore 路径（`write-delay-seconds: 0`），因此每次 |
There was a problem hiding this comment.
Shouldn't this be reflected in the documentation of benchmark testing? A more suitable approach is to conduct it in PR
| * independent job/pipeline counters that were initialized before measurement. Counter setup, | ||
| * MapStore reload checks, result validation, and cleanup are not timed. | ||
| * | ||
| * <p>Each increment waits on the write-through file-backed MapStore WAL append, so durable sync |
There was a problem hiding this comment.
Do these really need to be reflected in the comments?
| pipeline.getInProgress().removeIf(cp -> cp.getCheckpointId() == checkpointId); | ||
| } | ||
|
|
||
| /** |
There was a problem hiding this comment.
How much impact does this have on CV?
|
First of all, thank you for picking up this task so quickly and submitting the PR.😄 I think we should complete the performance evidence chain before concluding that this issue has been resolved. For this kind of performance optimization, I suggest following this process:
It is also important to distinguish between reducing average latency and reducing CV. Lower average latency does not automatically prove that the variance problem has been solved.
The complete evidence chain should ideally be: Benchmark symptom With this evidence in the PR description, reviewers can understand the causal relationship, reproduce the experiment locally, and verify that the optimization actually addresses the original latency-variance/CV problem rather than only improving the average score. |
…noise (apache#12058) Co-authored-by: Cursor <cursoragent@cursor.com>
|
@nzw921rx Thanks for the detailed review — I’ve addressed the inline comments and updated the PR description with the evidence pack. Inline review feedback
Core fix retained
Evidence chain (before / after)Same suite (
Java 8
Java 11
Causal chain (wall profile, Java 11 Diagnostics): Caveats I’m being explicit about:
Happy to add cpu/lock flame graphs as well if you still want them on top of the wall profiles. |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
What Problem Does This PR Solve?
Issue #12058 reported high within-run latency variance (CV) on CheckpointStorageBenchmark.checkpointIdAtomicIncrement and checkpointOverviewIncrementalUpdate, both of which exercise the write-through IMap MapStore path of the file-backed checkpoint WAL (imap-storage-file, write-delay-seconds: 0, i.e. every checkpoint state write blocks on a durable WAL append). The author traced the hot path (FileMapStore.store → HdfsWriter.flush → FSDataOutputStream.hsync) and found HdfsWriter.flush() was performing up to three sync calls per append on the real-HDFS branch. Fixing that is the headline change, but while investigating the wait path the author also found — and fixed — two independent, more serious correctness bugs in RequestFuture/WALWorkHandler that this PR bundles in. I want to be upfront that, having traced the code myself, the two bundled correctness fixes are actually more consequential for production safety than the sync-count reduction that the PR title advertises, and I've reviewed all three with equal weight below.
1. Code Change Review
1.1 Core Logic Analysis
Finding A — HdfsWriter.flush(): redundant syncs (the advertised fix)
Before:
public void flush() throws IOException {
if (out instanceof HdfsDataOutputStream) {
((HdfsDataOutputStream) out).hsync(EnumSet.of(SyncFlag.UPDATE_LENGTH));
}
if (out.getWrappedStream() instanceof DFSOutputStream) {
((DFSOutputStream) out.getWrappedStream()).hsync(EnumSet.of(SyncFlag.UPDATE_LENGTH));
} else {
out.hsync();
}
this.out.hflush();
}I traced this by hand for the real-HDFS case (out is an HdfsDataOutputStream wrapping a DFSOutputStream, the common production configuration): the first if has no return, so control falls through unconditionally into the second if, whose condition (getWrappedStream() instanceof DFSOutputStream) is also true — so hsync(UPDATE_LENGTH) is called a second time — and then hflush() runs unconditionally as a third call. This confirms the PR's claim of "up to three syncs on the HDFS path" from source, not just from the benchmark numbers.
For the local-file:///-backed benchmark path used in this PR's own evidence (out is a plain FSDataOutputStream, not HdfsDataOutputStream, and its wrapped stream is not a DFSOutputStream): the first if is skipped, the second if's else branch runs out.hsync(), then hflush() still runs unconditionally — 2 calls, not 3. This is consistent with the observed ~30% mean-latency drop (160.9→111.2 us/op on Java 8) being real but smaller than what a genuine HDFS cluster would show (3→1 calls there).
After: each of the three branches now ends with return, so exactly one sync call happens per flush(), always the strongest one available for that stream type.
Durability analysis (this is the part that actually matters for a checkpoint WAL): per the Hadoop Syncable contract, hsync() is documented as a strict superset of hflush() — it does everything hflush() does (push client buffer to the datanode pipeline) plus an additional durability guarantee (force to the physical device, and with UPDATE_LENGTH, persist the updated file length to the NameNode). Calling hflush() after hsync() therefore adds no additional durability — it's provably a no-op with respect to what's already been guaranteed. Removing it is a correct simplification, not a durability regression. I verified this isn't just asserted but actually tested: HdfsWriterDurableFlushTest.writeShouldPersistRecordsWithSingleSyncPath writes 8 records to a real (non-mocked) LocalFileSystem-backed WAL file and re-opens/re-reads the file with a fresh WALReader mid-stream (after record 4 and record 8, before close()), asserting every previously-flushed record is fully visible each time. That's the right test for "did we lose durability by removing a sync call" — it doesn't just check "no exception," it checks a fresh reader sees exactly what was flushed so far.
Finding B — RequestFuture: Future contract violations that silently discarded slow-but-legitimate writes (unadvertised, higher severity)
I want to flag this clearly because the PR title undersells it: this is a correctness fix for a bug that could cause the checkpoint state store to spuriously treat successful (but slow) WAL appends as failures.
Before:
public boolean isDone() { return success; } // conflates "done" with "succeeded"
public Boolean get() throws InterruptedException {
if (success) return true;
latch.await(1, TimeUnit.SECONDS); // hard-coded 1s cap, ignoring any caller-intended wait
if (!success) return false;
return success;
}
public Boolean get(long timeout, TimeUnit unit) throws InterruptedException {
if (success) return true;
latch.await(timeout, unit);
return success; // silently returns false on timeout instead of throwing
}I traced every production call site of RequestFuture in IMapFileStorage:
queryExecuteStatus(used by single-keystore/delete) already called the timedget(timeout, unit)with the real configuredwritDataTimeoutMilliseconds— so single-key writes were not directly hit by the 1-second cap.batchQueryExecuteFailsStatus(used bystoreAll/deleteAll, i.e. checkpoint overview batch writes — almost certainly the heavier, more representative write path forcheckpointOverviewIncrementalUpdate) called the bare, no-argrequestFuture.get(). That bareget()was hard-capped at exactly 1 second, regardless of the configured write timeout, whose default isDEFAULT_WRITE_DATA_TIMEOUT_MILLISECONDS = 1000 * 60(60 seconds) (IMapFileStorage.java:108). This is a 60x discrepancy between the configured tolerance and what was actually enforced for the batch path.- Consequence: any WAL append batch that legitimately took longer than 1 second — plausible under GC pause, disk contention, or exactly the kind of load this PR is trying to reduce variance under — would have that key added to the returned
failuresset (interpreted by the HazelcastMapStorebatch-write contract as "this entry needs to be retried/handled as failed"), even though the append was still in flight and would very likely have completed successfully moments later. The oldfinally { RequestFutureCache.remove(...) }also means that once the 1s timeout elapsed, nothing is left waiting for the actual completion — the disk write still happens, but its result is orphaned. This directly produces spurious retries/failures under exactly the load conditions the issue is about, which is the opposite of the intended reliability property of a "strict"/write-through state store.
After: isDone() now correctly reports latch.getCount() == 0L (completion, independent of success/failure — matching the Future interface contract, where isDone() must return true for a task that completed by success, exception, or cancellation). The no-arg get() now blocks until actually done (per Future contract — and I confirmed no production code calls the bare get() anymore, only a comment references it, so the now-indefinite block is safe). The timed get(timeout, unit) now throws TimeoutException on expiry instead of silently returning false — also the correct Future contract behavior. batchQueryExecuteFailsStatus was changed to pass this.writDataTimeoutMilliseconds explicitly (the same timeout single-key writes already used), closing the 60x discrepancy. All current call sites wrap the call in catch (Exception e) { log.error(...); }, so the new TimeoutException is caught and treated the same as the old silent-false — with the added benefit of a log line where previously a timeout was silently indistinguishable from a genuine failure. success was also changed from a plain boolean to a volatile boolean; this specifically matters for isDone()'s old return success form, which read the field without going through the CountDownLatch's happens-before edge — a real (if narrow) visibility gap that volatile closes (the get() paths were already safe via the latch's JMM guarantees, with or without volatile).
Finding C — WALWorkHandler: narrow catch that could permanently wedge checkpoint persistence (unadvertised, highest severity)
try {
writer.write(iMapFileData);
} catch (IOException e) { // BEFORE: only IOException caught
writeSuccess = false;
...
}
executeResponse(requestId, writeSuccess);WALWorkHandler.onEvent is the single LMAX Disruptor WorkHandler consuming every WAL append for this state store instance (single-threaded by design, per the class/module name). If writer.write(...) threw any unchecked exception that isn't an IOException — e.g. from a NullPointerException in a serializer, or an unchecked exception surfaced from within the Hadoop client stack — the old code let it propagate out of onEvent(). In LMAX Disruptor's default configuration, an uncaught exception escaping a WorkHandler terminates that worker thread (unless a custom ExceptionHandler was wired in, which I did not find evidence of here). Since this is the only consumer of the WAL ring buffer, that thread dying means every subsequent WAL append request is published but never processed again — every future RequestFuture for every future checkpoint write would sit unresolved until its configured timeout, then fail, forever, until the process restarts. That is a silent, cascading, single-poison-event outage of checkpoint persistence for the remainder of the JVM's life — exactly the class of P0 stability risk (thread death taking down an entire subsystem without crashing the JVM) that warrants top billing in a review, not a footnote.
After: the catch is widened to Exception (correctly still excluding Error, so genuinely fatal JVM conditions like OutOfMemoryError are not masked), and the accompanying comment states the reasoning precisely. executeResponse(requestId, writeSuccess) is now guaranteed to run for every APPEND event regardless of failure type, and — critically — the worker thread's onEvent() always returns normally, so the single consumer thread keeps running and keeps draining the ring buffer for all subsequent (unrelated) checkpoint writes. This is the right fix and the right catch granularity.
Runtime path (write-through checkpoint state store write):
FileMapStore.store()/storeAll()
-> IMapFileStorage.store()/storeAll()
-> sendToDisruptorQueue() [publish APPEND to ring buffer, register RequestFuture]
-> queryExecuteStatus()/batchQueryExecuteFailsStatus()
-> RequestFuture.get(timeout, unit) [blocks the calling checkpoint thread]
(disruptor thread) WALWorkHandler.onEvent()
-> writer.write() -> HdfsWriter.write() -> flush() [exactly one hsync now]
-> executeResponse() -> RequestFuture.done(success) [always reached now, even on non-IOException failure]
This is squarely on the normal, every-checkpoint production path for the file-backed WAL state store in write-through mode — not a boundary/recovery-only path.
1.2 Compatibility Impact
Fully compatible, and I independently verified the author's "No" answer to the user-facing-change question:
- No config option, default, public API, or wire/serialization format changes.
RequestFutureimplements the standardjava.util.concurrent.Future<Boolean>interface; the newget/get(timeout,unit)/isDone()behavior is a move toward contract compliance (an interface consumers are entitled to assume), not away from it.- Checkpoint/savepoint recovery semantics are untouched — this PR only affects how a single WAL append's completion is signaled and how many redundant sync syscalls happen per append, not the WAL file format, the recovery reader (
WALReader), or checkpoint metadata structure. - I confirm
TimeoutExceptionis a new checked exception onRequestFuture.get(timeout, unit)(matching the declaredFutureinterface, which already declaresthrows TimeoutException— so this isn't even a signature change, just an implementation that now actually honors the interface it always claimed to implement), and every current caller already catchesExceptionaround it, so no new uncaught-exception surface is introduced.
1.3 Performance / Side-Effect Analysis
- The
HdfsWriter.flush()change removes 1–2 redundant syscalls per WAL append on the write-through path — a straightforward, low-risk win with no new allocations, threads, or locks. - The
WALWorkHandlercatch-widening has no performance cost; it only affects the failure path. - The
RequestFuturechange fixes a case where the batch path was effectively enforcing a much shorter timeout than configured — closing that gap means batch writes may now legitimately wait longer (up to the real configured timeout) before being reported as failed, which is intentional and correct, not a regression; a caller who was inadvertently relying on the old accidental 1-second cap to "fail fast" was relying on a bug, not a documented behavior. - Evidence provided in the PR/issue thread (fork Actions
Benchmarksworkflow, same commit pair before/after, Java 8 and Java 11) shows both mean latency and CV improving on both benchmarks, with the author explicitly and honestly noting that GitHub-hosted runner CPU heterogeneity across runs makes absoluteScoreobservational rather than a hard regression gate — I agree with that caveat and would not want to see it dropped in future evidence packs.
1.4 Error Handling and Logging
WALWorkHandler's widened catch still logs"write orc file error, walEventBean is {}"with the exception at ERROR level (unchanged log statement, just now reachable for a broader exception set) — no swallowed exceptions, appropriate level for a failed durable write.RequestFuture.get(timeout, unit)throwingTimeoutException(instead of silently returningfalse) combined with the existingcatch (Exception e) { log.error("wait for write status error", e); }at bothIMapFileStoragecall sites means a timeout is now visible in logs where it previously vanished silently — a genuine observability improvement for exactly the kind of latency-variance investigation this issue is about.- No sensitive data newly logged.
2. Code Quality Assessment
2.1 Coding Standards
HdfsWriter.flush()now has a multi-line Javadoc explaining the "exactly one sync path" invariant and why redundant syncs don't add durability — this is exactly the kind of comment the project's rules require for a lifecycle/durability-critical method, and it's accurate.RequestFuturegained a class-level Javadoc explicitly warning thatisDone()reports completion, not success, and that callers must inspectget()— a good, non-obvious clarification given the class previously conflated the two.WALWorkHandler's catch-widening has an inline comment explaining the RuntimeException-kills-the-worker-thread rationale — accurate and appropriately placed at a non-trivial, easy-to-regress spot.successfield'svolatileaddition has no accompanying comment explaining why it's needed (i.e., thatisDone()reads it outside the latch's happens-before edge); a one-line comment here would help a future maintainer who might otherwise "clean up" what looks like an unnecessary modifier. Minor, non-blocking.
2.2 Test Coverage and Test Stability
RequestFutureTest(4 tests) covers: success completion, failure completion (making sureisDone()is now decoupled fromsuccess), timeout-throws-TimeoutExceptionwhile incomplete, and cross-thread completion observed by a waiting thread. All are deterministic — noThread.sleep-based polling, no shared static state, generous timeouts (2s) for the one cross-thread test relative to what it actually needs (adone()call from an already-started thread), and a fast (10ms) deterministic-by-construction timeout test (the future is simply never completed, so the timeout is guaranteed, not raced). Stability: Stable.HdfsWriterDurableFlushTest(1 test,@EnabledOnOs({LINUX, MAC})consistent with existing WAL test conventions forHADOOP_HOME/native-lib requirements) exercises the realLocalFileSystemwrite+flush+read-back path with mid-stream durability assertions, as analyzed in 1.1 Finding A. Deterministic, no timing assumptions. Stability: Stable.- Coverage gap (non-blocking, Low severity): the
HdfsDataOutputStream/wrapped-DFSOutputStreambranches offlush()— the ones with the actual "three syncs" bug on a real HDFS cluster — aren't exercised by any test here (understandably, since that would need aMiniDFSClusteror equivalent, heavier than this module's existing test style). I verified the fix is correct by code inspection (matchingSyncablecontract semantics per 1.1), but a future maintainer touching this method again won't have a test catching a regression specific to those two branches. - I note the author's response to
@nzw921rx's review correctly walked back the initially-includeddocs/zh|enbenchmark narrative, extra Javadoc, and an unrelatedcalculateStateSizestream→loop micro-optimization after being asked to keep the change focused — I checked the current diff and confirmed none of those are present; the PR is now scoped to exactly the WAL-sync/RequestFuture/WALWorkHandlerfix plus its two new test classes, which I agree is the right scope.
2.3 Documentation Updates
None needed — no user-facing config/API/behavior changed (confirmed independently in 1.2), consistent with the PR checklist.
3. Architectural Soundness
3.1 Elegance of the Solution
Precise fix for all three findings — each is a minimal, targeted change at the actual root cause (redundant flush() branches; Future-contract violations in RequestFuture; overly-narrow catch in the single WAL consumer), not a workaround. I'd specifically commend Finding C (WALWorkHandler) and Finding B (RequestFuture's timeout handling) as the kind of fix that's easy to miss while only looking at benchmark numbers — they were found by tracing the actual wait path, which is exactly the process @nzw921rx's review asked for, and the author's evidence response shows that process was genuinely followed (root-cause hypothesis → targeted change → before/after benchmark → durability test), not asserted after the fact.
3.2 Maintainability
All three files are more correct and more readable after this change (fewer branches in flush(), RequestFuture now matches the interface it implements, WALWorkHandler's failure handling is simpler to reason about). No new abstractions or indirection introduced.
3.3 Extensibility
No architectural impact — this is a bug fix within existing structures, not a new extension point.
3.4 Historical-Version Compatibility
No checkpoint/savepoint format, WAL file format, or recovery-path change. WALReader (the recovery-time reader) is untouched. A checkpoint written before this change and read back after it (or vice versa during a rolling restart) is unaffected, since the WAL file's on-disk byte layout (WALDataUtils.wrapperBytes) is not touched by this PR — only how many times the already-written bytes get sync()-ed, and how completion is signaled to the in-process caller.
4. Issue Summary
| # | Issue | Location | Severity |
|---|---|---|---|
| 1 | HdfsDataOutputStream/wrapped-DFSOutputStream sync-path branches (the real-HDFS "three syncs" scenario this PR targets) have no dedicated test; only the local-filesystem fallback branch is covered |
HdfsWriter.java:64-73, no corresponding test |
Low |
| 2 | success field's new volatile qualifier has no comment explaining it closes a real (if narrow) visibility gap in isDone(); a future reader may see it as decorative and remove it |
RequestFuture.java |
Low |
5. Merge Recommendation
Conclusion: Ready to merge
I respectfully disagree with treating @nzw921rx's comments as still-blocking — their review predates the author's revert of the out-of-scope doc/javadoc/micro-optimization changes and their explicit request for an evidence chain (root cause → profiling → hypothesis → targeted fix → before/after benchmark → durability verification) has been satisfied: the PR description and the follow-up issue comment provide before/after Score/Error/CV on both JDK 8 and 11 from the same fork-Actions benchmark workflow, name the exact call chain (FileMapStore.store → HdfsWriter.write/flush → FSDataOutputStream.hsync), and add a durability-focused test (HdfsWriterDurableFlushTest) specifically to verify the hsync/hflush behavior change doesn't compromise the property being changed. @nzw921rx's review state was "COMMENTED," not "CHANGES_REQUESTED," and I did not find any of the three original inline concerns (docs placement, extra Javadoc, CheckpointMonitorService CV impact) still present in the current diff — the first two were reverted, and I found no CheckpointMonitorService changes in the current diff at all, so that concern is moot.
- Blockers — none.
- Recommended fixes (non-blocking) — Issue 1 and Issue 2 above, both cosmetic/coverage nice-to-haves, not required before merge.
This PR is worth more than its title suggests. The advertised fix (collapsing HdfsWriter.flush() to one sync call) is correct and I verified it against the Syncable durability contract by hand, backed by a real (non-mocked) mid-stream durability test. But the two bundled fixes I'd ask reviewers to weight more heavily are: (1) RequestFuture's batch-write path was silently enforcing a 1-second timeout while the system was configured for 60 seconds, which could spuriously fail/retry legitimate slow-but-successful checkpoint batch writes under exactly the load conditions this issue describes; and (2) WALWorkHandler's narrow IOException catch could let an unchecked exception kill the sole WAL consumer thread and silently wedge all future checkpoint persistence for the life of the process. Both are now fixed correctly and are, in my view, more significant production-safety wins than the variance-reduction headline. Nice work tracing this all the way down to the actual wait path instead of stopping at the first plausible hotspot.
…e#12058) Co-authored-by: Cursor <cursoragent@cursor.com>
|
Thanks @DanielLeens for the thorough review — especially for independently tracing the HDFS multi-sync path, the On the two Low items:
Appreciate the merge recommendation and the re-weighting of the correctness fixes vs the CV headline. Happy to follow up with a MiniDFS-focused test in a separate PR if maintainers want that coverage next. |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Sorry for missing this in the last review — that's on me. I confidently described HdfsWriterDurableFlushTest as "the right test for did we lose durability by removing a sync call," but on a more critical re-read this round I don't think that claim fully holds up (details in 2.2 below). That test's content hasn't changed since the commit I reviewed last time (51bea001c), so this was discoverable then — it's a carryover miss, not something the new commit introduced. Thanks also to @Rangsh for e13bd21c8, which adds the one-line comment on volatile success I asked for (Issue 2 from last round) — that one is now closed.
What Problem Does This PR Solve?
Issue #12058 reports high within-run latency variance (CV ~15–19%) on CheckpointStorageBenchmark.checkpointIdAtomicIncrement and checkpointOverviewIncrementalUpdate. Both benchmarks exercise the write-through IMap MapStore backing SeaTunnel Zeta's engine_checkpoint-id-map and engine_checkpoint_monitor state stores — every checkpoint trigger has to synchronously wait for a durable WAL append to this file-backed store before the checkpoint coordinator can hand out the next checkpoint ID or update monitoring counters. Operators running with imap-storage-file and write-delay-seconds: 0 would see this as jittery per-checkpoint scheduling overhead that doesn't shrink even though the average is already small — the tail is what hurts.
The author traced the actual wait path (FileMapStore.store → IMapFileStorage.store → HdfsWriter.flush → FSDataOutputStream.hsync) instead of just staring at the benchmark, and the fix bundles three changes discovered along that trace: (1) HdfsWriter.flush() was calling up to three sync primitives per append with no early return, so this collapses it to exactly one, the strongest one available for the concrete stream type; (2) RequestFuture (the Future<Boolean> handle a caller blocks on while the WAL worker durably writes) had isDone()/get()/get(timeout,unit) implementations that violated the Future contract, including a hard-coded 1-second cap on the bare get() used by the batch write path even though the configured write timeout defaults to 60 seconds; (3) WALWorkHandler, the single LMAX Disruptor consumer thread for the whole WAL, only caught IOException around the write, so any unchecked exception would kill that thread and silently wedge all future checkpoint persistence for the rest of the process's life.
One-sentence summary: this collapses redundant WAL syncs to cut latency variance as advertised, and along the way fixes two independent, more serious bugs (a batch-write timeout that was 60x tighter than configured, and a single point of failure that could permanently stall checkpoint persistence) that matter more for production durability than the headline change does.
1. Code Change Review
1.1 Core Logic Analysis
Before/after example. Concretely, for one checkpoint trigger on a real HDFS-backed cluster, before this PR: HdfsWriter.flush() runs hsync(UPDATE_LENGTH) on the HdfsDataOutputStream branch, then falls through (no return) into the DFSOutputStream branch and runs hsync(UPDATE_LENGTH) again, then unconditionally runs hflush() — three sync-family calls for one durable append. After this PR, the first branch's hsync(UPDATE_LENGTH) call is immediately followed by return, so exactly one sync call happens. On the file:///-backed benchmark/test path (no HdfsDataOutputStream, no wrapped DFSOutputStream), before was out.hsync() then out.hflush() (2 calls); after is out.hsync() alone (1 call). I hand-verified this fall-through bug by reading HdfsWriter.java:68-80 (pre-PR at git show origin/dev:...HdfsWriter.java) myself — the missing return is real, not just asserted in the PR description.
Where this WAL actually sits (important scope correction vs. the PR title). I want to be precise here because "checkpoint state-store" is easy to misread as "the per-task checkpoint snapshot bytes." It is not. I traced EngineStateStoreNames (seatunnel-engine-server/.../common/statestore/EngineStateStoreNames.java:29-32): the imap-storage-file module this PR touches backs exactly four Hazelcast IMaps — engine_checkpoint-id-map, engine_error-handler-counter-map, engine_runningJobMetrics (explicitly no-op'd for persistence in FileMapStore.init, .../persistence/FileMapStore.java:46-53), and engine_checkpoint_monitor. The actual per-task checkpoint/savepoint state snapshot bytes go through a completely separate module family (seatunnel-engine-storage/checkpoint-storage-api + checkpoint-storage-plugins/checkpoint-storage-hdfs / checkpoint-storage-local-file), which shares no classes with what this PR touches (I grepped for IMapFileStorage|WALWorkHandler|RequestFuture|HdfsWriter under checkpoint-storage-plugins — zero hits) and is untouched by this diff. So this PR does not touch the barrier-injection → task-snapshot → checkpoint-storage-commit path at all. What it does touch is still load-bearing, though: StateStoreCheckpointIDCounter (seatunnel-engine-server/.../checkpoint/StateStoreCheckpointIDCounter.java:41-102) uses this exact store for getAndIncrement()/setCount(), and its own Javadoc says "the active coordinator, failover recovery, and savepoint restore path all resolve the same per-pipeline checkpoint sequence" through this counter. If an increment here were lost after being acknowledged, a new master after failover could hand out an already-used checkpoint ID, and since checkpoint IDs key the (separate) checkpoint-storage files, that's a real collision/overwrite risk on the actual snapshot data — just once removed from the WAL this PR changes, not directly in it.
Runtime flow — which step(s) this PR changes:
Checkpoint trigger (Master)
-> CheckpointIDCounter.getAndIncrement() / CheckpointMonitorService update
-> StateStoreCheckpointIDCounter -> CounterStateStore -> Hazelcast IMap put/replace
-> FileMapStore.store() [write-through MapStore, write-delay-seconds=0]
-> IMapFileStorage.store()/storeAll()
-> sendToDisruptorQueue() [publish APPEND, register RequestFuture]
-> queryExecuteStatus()/batchQueryExecuteFailsStatus()
-> RequestFuture.get(timeout, unit) <== CHANGED (Finding B: timeout now honors config)
(single disruptor thread) WALWorkHandler.onEvent()
-> writer.write() -> HdfsWriter.write() -> flush() <== CHANGED (Finding A: 1 sync, not up to 3)
-> executeResponse() -> RequestFuture.done() <== CHANGED (Finding C: catches Exception, not just IOException)
-- everything below is a SEPARATE, untouched code path --
-> barrier injected into task DAG -> task state snapshot
-> CheckpointStorage SPI (checkpoint-storage-hdfs / -local-file) commit
-> coordinator collects task acks -> checkpoint marked COMPLETED -> ack to client
Exactly three methods change: HdfsWriter.flush(), RequestFuture (isDone/get/get(timeout,unit)), and WALWorkHandler.onEvent()'s catch clause, plus IMapFileStorage's two call sites that had to change to match RequestFuture's corrected contract.
Durability of the sync change (the central question). Per the Hadoop Syncable interface contract, hsync() is documented as a strict superset of hflush(): it does everything hflush() does (push to the pipeline / visible to new readers) plus forcing the write to the physical device. Calling hflush() after hsync() therefore adds nothing — I traced every one of the three branches in the new flush() (HdfsWriter.java:68-80) and confirmed each of them still ends in exactly one hsync-family call before returning; none of the three branches was weakened to hflush()-only. I also confirmed write(byte[]) (HdfsWriter.java:82-86) calls this.flush() unconditionally and synchronously on every append — there is no code path where a WAL append returns to the caller (i.e., WALWorkHandler calls executeResponse()/RequestFuture.done()) without that single sync having already completed. Since RequestFuture.done() only runs after writer.write() returns (WALWorkHandler.java:70-76), the sequencing that matters — sync-before-ack — is intact on every path, savepoint included (savepoint restore resolves the same counter store per StateStoreCheckpointIDCounter's own Javadoc, so there's no separate, unsynced fast path for it).
Concurrency. WALWorkHandler is the sole LMAX Disruptor WorkHandler for this WAL (class doc: "Single thread to write data to orc file"), so appends were already serialized before this PR by the single-consumer design, independent of how many sync calls each append made. Removing redundant syncs doesn't remove any serialization the syncs themselves were providing — the Disruptor's own dispatch is what serializes concurrent checkpoint writers, and that's unchanged.
Finding B verified independently. I confirmed IMapFileStorage.java:314-328 (queryExecuteStatus, single-key path) always passed the real configured timeout even before this PR, but batchQueryExecuteFailsStatus (IMapFileStorage.java:330-353, used by storeAll/deleteAll) called the bare RequestFuture.get(), which was hard-capped at 1 second regardless of writDataTimeoutMilliseconds (default 60,000 ms, IMapFileStorage.java:108). After this PR it passes this.writDataTimeoutMilliseconds explicitly, closing that 60x gap. I grepped every caller of requestFuture.get in seatunnel-engine and found exactly these two, both now using the timed overload — the previously-unbounded-block risk from making the bare get() wait on latch.await() forever is moot because nothing in production calls it anymore.
Finding C verified independently. WALWorkHandler.java:70-75 now catches Exception (not Error, correctly), and executeResponse() is unconditionally reached either way, so the single consumer thread's onEvent() never propagates an exception that would kill it under Disruptor's default WorkHandler semantics.
1.2 Compatibility Impact
Fully compatible. No config option, default, public API, or on-disk WAL/checkpoint format changes. WALDataUtils.wrapperBytes (the WAL record byte layout) and WALReader (the recovery-time reader) are both untouched — I confirmed via git diff origin/dev...pr-12081 --stat that only IMapFileStorage.java, WALWorkHandler.java, RequestFuture.java, HdfsWriter.java, and two new test files changed; nothing under common/WALDataUtils.java or common/WALReader.java appears in the diff. RequestFuture implements Future<Boolean>; the new behavior moves strictly toward the interface's documented contract, and get(timeout, unit)'s new TimeoutException is already declared by Future (not a new checked-exception surface) and already caught by both call sites' catch (Exception e). A checkpoint/savepoint written before this change and read back after (or a rolling-upgrade in-flight checkpoint) is unaffected — this only changes how many syscalls happen per already-identical byte-for-byte append and how completion is signaled in-process.
1.3 Performance / Side-Effect Analysis
The flush() change removes 1–2 redundant syscalls per WAL append with no new allocation/locking. The WALWorkHandler catch-widening only affects the failure path, no steady-state cost. The RequestFuture timeout fix is a legitimate behavior change I want to call out explicitly as a side effect, not a regression: batch writes can now legitimately wait up to the real configured timeout (60s default) instead of failing fast at 1s. A caller that depended on the old 1-second cap to fail fast was depending on a bug (the batch path silently enforcing 1/60th of the documented timeout), not a documented property — closing that gap is correct, but operators upgrading should know that a genuinely wedged WAL worker (not fixed by this PR, e.g. permanent HDFS unavailability) will now surface as a much slower failure (up to 60s) than before (1s). Given WALWorkHandler's catch-widening in the same PR makes the "worker thread dies and nothing ever completes" scenario far less likely, I don't think this is a net regression, but it's worth the author/maintainers being aware of it as a documented trade-off rather than a free lunch.
On the evidence: the author's benchmark tables (Java 8/11, same fork Actions Benchmarks workflow, before commit 98182ca59 vs. after fa12fbd60) show both mean latency and CV improving on both benchmarks (e.g. checkpointIdAtomicIncrement Java 8: 160.9→111.2 us/op, CV 14.59%→9.62%). The reduction in sync-family syscalls plausibly explains a CV drop (syscall latency itself has variance under contention, especially the redundant second/third call), and the direction is consistent across both JDKs, which I find persuasive even with the author's own honest caveat that absolute Score is muddied by heterogeneous GitHub-hosted runner CPUs. On "could batching achieve the same without removing a sync": yes, group-commit batching (coalescing many callers' WAL appends behind one shared flush) is a legitimate alternative technique, but it would require batching multiple concurrent RequestFutures behind one flush call with its own partial-failure semantics — a materially larger, riskier design change than "stop calling sync twice for one already-durable write." I'd file batching as a reasonable follow-up idea, not a reason to block this narrower, lower-risk fix.
1.4 Error Handling and Logging
WALWorkHandler's widened catch still logs at ERROR with the exception ("write orc file error, walEventBean is {}", WALWorkHandler.java:74) for every failure type now caught — nothing is swallowed silently. RequestFuture.get(timeout, unit) throwing TimeoutException instead of silently returning false means a timeout is now visible via the existing catch (Exception e) { log.error("wait for write status error", e); } at both IMapFileStorage call sites (IMapFileStorage.java:322-323, 343-344) — a genuine observability improvement for exactly the kind of variance investigation issue #12058 describes. No sensitive data newly logged.
2. Code Quality Assessment
2.1 Coding Standards
HdfsWriter.flush() (HdfsWriter.java:61-67) and RequestFuture's class Javadoc (RequestFuture.java:28-33) are accurate, multi-line Javadoc that explain the non-obvious invariant (why redundant syncs don't add durability; that isDone() reports completion not success) — exactly the kind of comment the project's rules require for a durability-critical method. WALWorkHandler's catch-widening has an inline comment explaining the Disruptor worker-thread-death rationale. The volatile success field now has its explanatory comment too (RequestFuture.java:38-39, added in e13bd21c8), closing the one item I flagged last round.
2.2 Test Coverage and Test Stability
RequestFutureTest (.../future/RequestFutureTest.java, 4 tests, lines 33-77) covers success completion, failure-without-conflating-with-completion, timeout-throws-TimeoutException, and cross-thread completion. All are deterministic: no Thread.sleep-based polling, generous fixed timeouts (2s at line 73 for a thread that starts immediately, 10ms at line 57 for a future that's never completed so the timeout is guaranteed by construction). Stability: Stable.
HdfsWriterDurableFlushTest (.../wal/writer/HdfsWriterDurableFlushTest.java, 1 test, lines 54-96) is deterministic in its own right — no timing assumptions, @EnabledOnOs({LINUX, MAC}) consistent with existing WAL test conventions. Stability: Stable as a test (it will not flake).
However, on the specific question of whether it proves what's claimed — and this is the carryover point I opened with — I don't think it does, fully. It writes via HdfsWriter, then reads back mid-stream through a second WALReader handle in the same process, same live OS (lines 79-85, 91-95) and asserts the records are visible. That's a valid test of read-your-own-write visibility across handles, which is what hflush() already guarantees. But the specific extra guarantee hsync() is supposed to add over hflush(), per the same Syncable contract I cited in 1.1, is durability to the physical device — i.e., survives an OS crash or power loss, not just "another live file handle on the same machine can see it." A second reader on the same live OS will see hflush()-only data too, because that data is sitting in the OS page cache regardless of whether it was ever fsynced to disk — page cache is shared across file descriptors on the same running kernel. So this test cannot, by construction, distinguish "we correctly kept calling hsync()" from a hypothetical regression where the collapsed flush() accidentally called only hflush()-equivalent behavior on some branch — both would pass this test identically. It's a good regression test for "we didn't break write-then-read-back ordering," which does matter, but it is not a crash-survival proof, and I was wrong to describe it as one last round.
To be clear about where this leaves the correctness conclusion: I still believe Finding A is correct, but that conclusion rests on the documented Syncable.hsync()/hflush() contract plus my own code trace (every one of the three flush() branches still ends in an hsync-family call, none was weakened) — not on this test. That's a legitimate way to establish correctness for a well-specified public interface, so I'm not treating this as a blocker; a mocked-hsync()/spy-based unit test that asserts the previously-hflush()-only path (the third flush() branch, exercised by this very test's LocalFileSystem setup) actually still invokes an hsync-family method — rather than only checking downstream visibility — would close this gap cheaply and I'd ask for it as a follow-up.
Separately, and consistent with the still-open item from last round: the HdfsDataOutputStream/wrapped-DFSOutputStream branches — the ones with the actual "three syncs" bug on a real HDFS cluster — remain untested (would need MiniDFSCluster or equivalent). The author's response in the issue thread explains this is a deliberate scope decision, offering a MiniDFS-focused follow-up PR; I think that's a reasonable trade-off given the module has no existing MiniDFS harness, so I'm not blocking on it, same as last round.
2.3 Documentation Updates
None needed — confirmed independently, no user-facing config/API/behavior changed, and the earlier docs/zh/docs/en benchmark-narrative addition (flagged by @nzw921rx) was reverted in 51bea001c and is not present in the current diff.
3. Architectural Soundness
3.1 Elegance of the Solution
Each of the three findings is a minimal, targeted fix at its actual root cause — a missing return, a Future implementation that didn't honor its own interface, and an overly narrow catch. None of them is a workaround or a new abstraction layered on top of the problem.
3.2 Maintainability
All three files are more correct and easier to reason about after this change (fewer branches in flush(), RequestFuture now matches the interface it declares, WALWorkHandler's failure handling is simpler). No new indirection introduced.
3.3 Extensibility
No architectural impact — this is a bug fix within existing structures, not a new extension point.
3.4 Historical-Version Compatibility
Fully compatible. No checkpoint/savepoint format, WAL file format, or recovery-path change — WALReader and WALDataUtils are untouched by this diff, and I independently confirmed that via the file-level diff stat. A checkpoint-id-map entry (or checkpoint-monitor entry) written by an old binary and read by a new one, or vice versa during a rolling upgrade, is unaffected: the bytes on disk are identical, only the number of sync syscalls per append and the in-process completion-signaling semantics changed.
4. Issue Summary
| # | Issue | Location | Severity |
|---|---|---|---|
| 1 | HdfsWriterDurableFlushTest proves cross-handle same-process visibility, not true fsync-survives-a-crash durability (page cache is shared across handles on the same live OS regardless of hsync vs hflush); the correctness conclusion for Finding A rests on the documented Syncable contract plus manual code trace, not on this test |
.../wal/writer/HdfsWriterDurableFlushTest.java:79-95 |
Medium |
| 2 | HdfsDataOutputStream/wrapped-DFSOutputStream sync-path branches (the real-HDFS "three syncs" scenario this PR targets) still have no dedicated test; only the local-filesystem fallback branch is covered (carried over from last round, author has explained the scope decision) |
HdfsWriter.java:69-79, no corresponding test |
Low |
5. Merge Recommendation
Conclusion: Ready to merge
No High-severity blockers. I traced every branch of the changed flush() method, every caller of RequestFuture.get, and the single-consumer Disruptor model, and I'm satisfied the sync-before-ack sequencing that matters for durability is intact on every path this PR touches, including the checkpoint-ID-counter path that failover/savepoint restore depends on. The two bundled correctness fixes (RequestFuture's batch-timeout contract violation, WALWorkHandler's narrow catch) are real, verified independently against production call sites, and matter more for production safety than the headline variance reduction.
The one thing keeping this from an unqualified "verified end to end" is Issue 1 above: the new durability test doesn't actually prove what I previously (incorrectly) credited it with proving. I don't think that rises to a blocker, because the underlying correctness claim is still provable by contract + code trace independent of that test, and I'd rather say so plainly than inflate a real-but-non-blocking gap into a rejection. That said, this is a checkpoint-adjacent durability change, so I'd like the crash-durability question closed properly rather than left as a documented gap forever.
- Blockers — none.
- Recommended fixes (non-blocking):
- Issue 1: add (or replace) the durability test with one that verifies
hsync-family methods are actually invoked (e.g. a spy/mock on the wrapped stream, or an assertion on call counts per branch) rather than only asserting downstream read visibility, so a future regression that silently downgrades a branch tohflush()-only would actually be caught. - Issue 2: a
MiniDFSCluster-backed test for the two real-HDFSflush()branches, as the author already offered to follow up with.
- Issue 1: add (or replace) the durability test with one that verifies
Nice work tracing this all the way to the actual wait path, and thanks for the quick turnaround on the volatile comment.
…che#12058) Co-authored-by: Cursor <cursoragent@cursor.com>
|
Thanks @DanielLeens for the careful re-read — and for correcting the earlier durability-test claim. That distinction between same-process read-back visibility ( What I changedIssue 1 (Medium): added
So a silent regression that downgrades a branch to Issue 2 (Low / MiniDFS): still treating a real Also noted your point on the batch-timeout side effect (effective 1s → configured 60s default): intentional contract fix, not a free lunch for operators who were relying on the old accidental fail-fast. Appreciate the thorough review again. |
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks @Rangsh for 70e40b40e5 and for the detailed writeup in your latest comment. I re-read the entire PR from scratch against the current head rather than only diffing since my last pass, and I confirm your description matches what I independently verified: the only change since the commit I last reviewed (e13bd21c8) is two test files — HdfsWriterDurableFlushTest.java gained a corrected class Javadoc, and a brand-new HdfsWriterFlushSyncPathTest.java was added. No production code changed (git compare e13bd21c8...70e40b40e5 touches exactly those two test files, +5/-2 and +98/-0). Issue 1 (Medium) from my last round — that the durability test only proved same-process visibility, not a real hsync invocation — is now closed by the new mock-based test; details below. Issue 2 (Low, MiniDFS coverage) remains open as a deliberate, previously-discussed follow-up. I also re-verified @nzw921rx's three original inline concerns (docs placement, extra Javadoc, CheckpointMonitorService CV impact) are still absent from the current diff, consistent with their revert in 51bea001c.
What Problem Does This PR Solve?
Issue #12058 reports high within-run latency variance (CV roughly 15-19%) on CheckpointStorageBenchmark.checkpointIdAtomicIncrement and checkpointOverviewIncrementalUpdate. Both benchmarks exercise the write-through IMap MapStore backing SeaTunnel Zeta's engine_checkpoint-id-map and engine_checkpoint_monitor state stores: with imap-storage-file and write-delay-seconds: 0, every checkpoint-ID increment and every checkpoint-monitor update synchronously blocks on a durable WAL append before the caller can proceed. Operators running this configuration see this as jittery per-checkpoint scheduling overhead that does not shrink even though the mean is already small.
The author traced the actual wait path (FileMapStore.store -> IMapFileStorage.store -> HdfsWriter.flush -> FSDataOutputStream/HdfsDataOutputStream sync calls) instead of stopping at the benchmark numbers, and bundled three fixes discovered along that trace: (1) HdfsWriter.flush() was missing early returns, so on a real HDFS cluster it issued up to three sync-family calls per append; this PR collapses it to exactly one, the strongest sync available for the concrete stream type. (2) RequestFuture, the Future<Boolean> handle a caller blocks on while the WAL worker durably writes, had isDone()/get()/get(timeout,unit) implementations that violated the Future contract, including a hard-coded 1-second cap on the bare get() used by the batch-write path even though the configured write timeout defaults to 60 seconds. (3) WALWorkHandler, the single LMAX Disruptor consumer thread for the whole WAL, only caught IOException around the write, so any unchecked exception could kill that thread and silently wedge all future checkpoint persistence for the rest of the process's life.
One-sentence summary: this PR collapses redundant WAL syncs to cut checkpoint-store latency variance as advertised, and along the way fixes two independent, more serious bugs — a batch-write timeout that was effectively 60x tighter than configured, and a single point of failure that could permanently stall checkpoint persistence — that matter more for production durability than the headline change does.
1. Code Change Review
1.1 Core Logic Analysis
Concrete before/after example. For one checkpoint-ID increment on a real HDFS-backed cluster: before this PR, HdfsWriter.flush() ran hsync(UPDATE_LENGTH) on the HdfsDataOutputStream branch (out instanceof HdfsDataOutputStream), then fell through with no return into the DFSOutputStream branch and ran hsync(UPDATE_LENGTH) a second time, then ran hflush() unconditionally as a third call. After this PR (HdfsWriter.java:68-80), each of the three branches ends in return immediately after its single sync call, so exactly one sync-family call happens per flush(). On the file:///-backed benchmark/test path used in the PR's own evidence, the count goes from 2 calls (out.hsync() then out.hflush()) to 1 (out.hsync() alone).
Where this WAL actually sits. The imap-storage-file module this PR touches backs four Hazelcast IMaps: engine_checkpoint-id-map, engine_error-handler-counter-map, engine_runningJobMetrics (explicitly not persisted, see FileMapStore.init), and engine_checkpoint_monitor. The per-task checkpoint/savepoint state snapshot bytes flow through a separate module family (checkpoint-storage-api / checkpoint-storage-plugins) that shares no classes with this diff. What this PR does touch is still load-bearing: StateStoreCheckpointIDCounter uses this exact store for getAndIncrement()/setCount(), and the active coordinator, failover recovery, and savepoint restore all resolve the same per-pipeline checkpoint sequence through this counter. So this PR is on the normal, every-checkpoint production path for the counter/monitor state, not a boundary-only path, even though it is one hop removed from the barrier-injection/task-snapshot commit path itself.
Durability of the sync change. Per the Hadoop Syncable contract, hsync() is a strict superset of hflush(): it does everything hflush() does plus forcing the write to the physical device (and, with UPDATE_LENGTH, persisting the updated length to the NameNode). Calling hflush() after hsync() therefore adds nothing. I traced all three branches in the current flush() (HdfsWriter.java:68-80) and confirmed each still ends in exactly one hsync-family call before returning; none was weakened to hflush()-only. write(byte[]) (HdfsWriter.java:82-86) calls this.flush() unconditionally and synchronously on every append, and RequestFuture.done() (WALWorkHandler.java:76, RequestFuture.java:71-74) only runs after writer.write() returns, so sync-before-ack is intact on every path, including the checkpoint-ID counter path that failover/restore depends on.
RequestFuture fix, verified against call sites. queryExecuteStatus (IMapFileStorage.java:318-328, single-key store/delete) already used the timed get(timeout, unit) with the real configured timeout before this PR. batchQueryExecuteFailsStatus (IMapFileStorage.java:330-352, used by storeAll/deleteAll, the batch path exercised by checkpointOverviewIncrementalUpdate) called the bare, no-arg get(), hard-capped at 1 second regardless of writDataTimeoutMilliseconds (default DEFAULT_WRITE_DATA_TIMEOUT_MILLISECONDS = 60_000, IMapFileStorage.java:108). A WAL append batch taking longer than 1 second under GC pause or disk contention would have its key added to failures even though the append was still in flight and likely to succeed moments later. After this PR, batchQueryExecuteFailsStatus explicitly passes this.writDataTimeoutMilliseconds (IMapFileStorage.java:339-342), and RequestFuture.isDone() now reports latch.getCount() == 0L (RequestFuture.java:53-55, matching the Future contract's completion-vs-success distinction), get() blocks until actually done (RequestFuture.java:58-61), and get(timeout, unit) throws TimeoutException on expiry instead of silently returning false (RequestFuture.java:64-69). I grepped every requestFuture.get call site in seatunnel-engine and confirmed both remaining callers use the timed overload and both already wrap the call in catch (Exception e), so the new checked TimeoutException (already declared by Future, not a new signature) is caught, not a new uncaught-exception surface.
WALWorkHandler fix. The single Disruptor WorkHandler for this WAL now catches Exception instead of only IOException (WALWorkHandler.java:70-75), with executeResponse(requestId, writeSuccess) unconditionally reached either way (WALWorkHandler.java:76). Error is correctly still left uncaught, so genuinely fatal JVM conditions are not masked. Since onEvent() never propagates now, the sole consumer thread never dies from an unchecked exception in writer.write(), which previously could permanently wedge all future checkpoint persistence for the process's remaining life under Disruptor's default WorkHandler semantics.
Concurrency. WALWorkHandler is the sole consumer for this WAL (single-thread-by-design), so appends were already serialized independent of sync-call count; removing redundant syncs does not remove any serialization, since the Disruptor's own dispatch — not the syscalls — provides it.
1.2 Compatibility Impact
Fully compatible. No config option, default, public API, or on-disk WAL/checkpoint format change. WALDataUtils.wrapperBytes (WAL record byte layout) and WALReader (the recovery-time reader) are both untouched by this diff. RequestFuture implements Future<Boolean>; the corrected behavior moves strictly toward the interface's documented contract. get(timeout, unit)'s TimeoutException is already declared by Future (not a new checked-exception surface) and already caught at both call sites. A checkpoint/savepoint written before this change and read back after (or an in-flight checkpoint during a rolling upgrade) is unaffected: the bytes on disk are byte-for-byte identical, only the syscall count per append and the in-process completion signaling changed.
1.3 Performance / Side-Effect Analysis
The flush() change removes 1-2 redundant syscalls per WAL append with no new allocation or locking. The WALWorkHandler catch-widening only affects the failure path. The RequestFuture fix is a real, intentional behavior change worth calling out as a trade-off rather than a free win: batch writes can now legitimately wait up to the real configured timeout (60s default) instead of failing fast at the old, accidental 1s cap. A caller relying on that 1-second cap to fail fast was relying on a bug (the batch path silently enforcing 1/60th of the documented timeout), not a documented property, so closing the gap is correct; but it does mean a genuinely wedged WAL worker now surfaces as a slower failure than before that fix alone, which is why bundling it with the WALWorkHandler catch-widening (which reduces how often the worker gets wedged in the first place) is the right pairing. The author's before/after benchmark evidence (fork Actions Benchmarks workflow, same commit pair, Java 8 and Java 11) shows both mean latency and CV improving on both benchmarks (for example checkpointIdAtomicIncrement on Java 8: 160.9 to 111.2 us/op, CV 14.59% to 9.62%), with an explicit and appropriate caveat that GitHub-hosted runner CPU heterogeneity makes the absolute Score observational rather than a hard regression gate.
1.4 Error Handling and Logging
WALWorkHandler's widened catch still logs at ERROR with the exception (WALWorkHandler.java:74) for every failure type now caught; nothing is swallowed. RequestFuture.get(timeout, unit) throwing TimeoutException instead of silently returning false means a timeout is now visible via the existing catch (Exception e) { log.error("wait for write status error", e); } at both IMapFileStorage call sites (IMapFileStorage.java:322-323, 343-344), a genuine observability improvement for exactly the kind of variance investigation issue #12058 describes. No sensitive data newly logged.
2. Code Quality Assessment
2.1 Coding Standards
HdfsWriter.flush() (HdfsWriter.java:61-67) and RequestFuture's class Javadoc (RequestFuture.java:28-33) are accurate multi-line Javadoc explaining the non-obvious invariants (why redundant syncs add no durability; that isDone() reports completion, not success). WALWorkHandler's catch-widening has an inline comment explaining the Disruptor worker-thread-death rationale. The volatile success field now has its explanatory comment (RequestFuture.java:38-39), closing the item I flagged in my first round.
2.2 Test Coverage and Test Stability
RequestFutureTest (4 tests) covers success completion, failure-without-conflating-with-completion, timeout-throws-TimeoutException, and cross-thread completion. All deterministic: no sleep-based polling, a 10ms timeout test that is guaranteed by construction (the future is simply never completed), a 2s timeout for a thread that starts immediately. Stability: Stable.
HdfsWriterFlushSyncPathTest.java is the new file added by 70e40b40e5 and it directly, correctly closes Issue 1 from my last round. I traced each of its three tests against the current flush() control flow:
flushShouldCallHdfsDataOutputStreamHsyncOnce(lines 49-59):outis a mockHdfsDataOutputStream, so the first branch (out instanceof HdfsDataOutputStream,HdfsWriter.java:69-73) is taken and returns immediately. Assertshsync(UPDATE_LENGTH)called exactly once, plainhsync()andhflush()never called.flushShouldCallWrappedDfsOutputStreamHsyncOnce(lines 61-76):outis a plainFSDataOutputStreammock whosegetWrappedStream()returns aDFSOutputStreammock, so the second branch (HdfsWriter.java:74-78) is taken. The test assertsgetWrappedStream()is called exactly twice (once for theinstanceofcheck, once for the cast), matching the actual source, and thatdfs.hsync(UPDATE_LENGTH)is called exactly once withhsync()/hflush()never called on either mock.flushShouldCallPlainFsDataOutputStreamHsyncOnce(lines 78-89): neither branch matches, so it falls through toout.hsync()(HdfsWriter.java:79). AssertsgetWrappedStream()called once,hsync()called once,hflush()never called.
This is exactly the spy/call-count style test I asked for: it fails if any branch silently regresses to hflush()-only or stacks a second sync call, which same-process read-back visibility (the previous test's only signal) could not detect. I verified the field-injection helper (writerWithOut, lines 91-97) matches the real private field name out in HdfsWriter, and that mock(HdfsDataOutputStream.class)/mock(FSDataOutputStream.class) are usable here because the root pom.xml declares mockito-junit-jupiter and mockito-inline as plain (non-managed) test-scope dependencies inherited by every module, so this module needed no pom change to compile the new test — I checked imap-storage-file/pom.xml and confirmed it declares no direct Mockito dependency, relying entirely on the inherited one, which is correct and matches how other modules in this codebase use Mockito. Stability: Stable — no timing, no shared state, deterministic mock verification.
HdfsWriterDurableFlushTest.java's class Javadoc (lines 45-53) is now correctly scoped: "This is not a crash-survival / fsync proof: same-process read-back also passes for hflush()-only data sitting in the OS page cache," with a cross-reference to the new test for the actual sync-invocation guarantee. This matches what I asked for exactly and is accurate; the test itself is unchanged and remains a valid (if narrower-scoped) regression test for write-then-read-back ordering. Stability: Stable.
Remaining gap (carried over from my last round, Low, not new): the HdfsDataOutputStream/wrapped-DFSOutputStream branches — the ones with the actual "three syncs" bug on a real HDFS cluster — still have no MiniDFSCluster-backed integration test. I want to be precise about what the new commit does and does not close here: HdfsWriterFlushSyncPathTest now exercises those two branches' control flow (call counts) via mocks, which is a real improvement over the previous state (only the local-filesystem fallback branch was covered at all), but it does not exercise the real Hadoop client stack end-to-end. The author has explained this is a deliberate scope decision given the module has no existing MiniDFSCluster harness, and has offered a follow-up PR. I agree that is a reasonable trade-off and do not consider it blocking, same conclusion as my last round, now on firmer ground given the mock-level coverage that now exists.
2.3 Documentation Updates
None needed. No user-facing config/API/behavior changed. The earlier docs/zh/docs/en benchmark-narrative addition flagged by @nzw921rx was reverted in 51bea001c and remains absent from the current diff, which I independently confirmed via the full PR diff file list (IMapFileStorage.java, WALWorkHandler.java, RequestFuture.java, HdfsWriter.java, and three test files only).
3. Architectural Soundness
3.1 Elegance of the Solution
Each of the three findings is a minimal, targeted fix at its actual root cause: a missing return, a Future implementation that did not honor its own interface, and an overly narrow catch. None is a workaround or a new abstraction layered on top of the problem. The new HdfsWriterFlushSyncPathTest is likewise a minimal, targeted addition that closes exactly the gap it was asked to close, without expanding scope.
3.2 Maintainability
All three production files are more correct and easier to reason about after this change: fewer branches in flush(), RequestFuture now matches the interface it declares, WALWorkHandler's failure handling is simpler. No new indirection introduced. The corrected Javadoc on HdfsWriterDurableFlushTest reduces the risk of a future reader over-trusting what that specific test proves.
3.3 Extensibility
No architectural impact. This is a bug fix within existing structures, not a new extension point.
3.4 Historical-Version Compatibility
Fully compatible. No checkpoint/savepoint format, WAL file format, or recovery-path change. WALReader and WALDataUtils are untouched by this diff. A checkpoint-id-map entry or checkpoint-monitor entry written by an old binary and read by a new one, or vice versa during a rolling upgrade, is unaffected: only the syscall count per already-identical append and the in-process completion signaling changed.
4. Issue Summary
| # | Issue | Location | Severity | Raised by another reviewer |
|---|---|---|---|---|
| 1 | HdfsDataOutputStream/wrapped-DFSOutputStream sync-path branches (the real-HDFS "three syncs" scenario this PR targets) still have no MiniDFSCluster-backed end-to-end test; only mock-level control-flow coverage exists as of 70e40b40e5 |
HdfsWriter.java:69-78, no MiniDFS test |
Low | No |
5. Merge Recommendation
Conclusion: Ready to merge
No blockers, in this round or carried over. Issue 1 from my last review (Medium: the durability test did not actually prove what I had previously credited it with proving) is closed by HdfsWriterFlushSyncPathTest in 70e40b40e5 — I traced all three of its tests against the current flush() branches line by line and confirmed each assertion matches the real control flow and would fail on the exact regression class it targets (a branch silently downgrading to hflush()-only, or stacking a second sync call). The companion Javadoc fix on HdfsWriterDurableFlushTest correctly narrows that test's claimed guarantee instead of leaving the old overclaim in place. No production code changed since my last review, so the durability, compatibility, and concurrency conclusions from that round stand unchanged and I re-verified them from scratch this round rather than merely re-citing them.
Note on CI: at the time of this review the apache-side "Build" check shows a cancelled conclusion, but the corresponding fork Actions run for this exact head commit (70e40b40e56459676f8f808a3a8747bbed413d7b) is still queued as of this writing, alongside an earlier cancelled run at the same timestamp that looks like a duplicate-trigger artifact rather than a real failure signal. I am not treating this as a blocker one way or the other — my conclusion above is based entirely on source-level analysis, consistent with this project's SeaTunnel local-validation policy, and the queued run should be allowed to complete on its own before it is read as pass or fail.
- Blockers — must be fixed: none.
- Recommended fixes — non-blocking:
- Issue 1: a
MiniDFSCluster-backed integration test for the two real-HDFSflush()branches, as already offered by the author as a follow-up PR.
- Issue 1: a
This PR is worth more than its title suggests, and that conclusion has not changed across two rounds of full re-review. The advertised fix (collapsing HdfsWriter.flush() to one sync call) is correct, verified against the Hadoop Syncable durability contract and now backed by a real call-count test rather than only an indirect visibility test. The two bundled fixes — RequestFuture's batch-write path silently enforcing a 60x-tighter timeout than configured, and WALWorkHandler's narrow catch that could permanently wedge the sole WAL consumer thread — remain, in my view, more significant production-safety wins than the variance-reduction headline. Good, precise response to review feedback across both rounds.
3 participants
Purpose of this pull request
Fixes #12058.
Investigate and reduce the within-run latency variance observed on:
CheckpointStorageBenchmark.checkpointIdAtomicIncrementCheckpointStorageBenchmark.checkpointOverviewIncrementalUpdateRoot cause:
write-delay-seconds: 0).HdfsWriter.flush()previously performed redundanthsync/hflushcalls (up to three syncs on the HDFS path), which increased latency and CV without improving durability.FileMapStore.store→HdfsWriter.flush→FSDataOutputStream.hsync.This PR applies a focused production-side fix while preserving checkpoint correctness and state-store durability:
HdfsWriter.flush().RequestFuturecompletion/success semantics and make batch WAL waits use the configured write timeout.WALWorkHandleralways publishesdone()for APPEND events, even on non-IOExceptionfailures.Out of scope after review (intentionally reverted):
docs/en|zhbenchmark guides (discussed in this PR description instead)CheckpointStorageBenchmarkjavadoccalculateStateSize/ overview stream→loop micro-optimizations (not the CV driver)Does this PR introduce any user-facing change?
No.
hsync.How was this patch tested?
Unit tests:
RequestFutureTestHazelcastCheckpointOverviewStateStoreTestHdfsWriterDurableFlushTest(enabled on Linux/macOS, matching existing imap-storage-file WAL tests; skipped on Windows whenHADOOP_HOMEis unset)Local commands:
Evidence (before / after)
Suite:
CheckpointStorageBenchmarkonly, via fork ActionsBenchmarksworkflow (GitHub-hosted ubuntu-24.04). Storage uses localfile:///(Hadoop LocalFileSystem); no remote HDFS cluster required.dev@98182ca5933840694352improve/zeta-checkpoint-state-store-latency-12058@fa12fbd6033840699925Java 8
checkpointIdAtomicIncrementcheckpointOverviewIncrementalUpdateJava 11
checkpointIdAtomicIncrementcheckpointOverviewIncrementalUpdateNotes:
Scoreis observational, not a hard regression gate; the directional Error/CV reduction is the primary signal for this issue.FileMapStore.store→HdfsWriter.write/flush→FSDataOutputStream.hsyncon the critical path for both methods.Check list
New License Guide
incompatible-changes.mdto describe the incompatibility caused by this PR.