Repository navigation
fix(offset): commit offsets in handoff-sequence FIFO order, never re-execute a written batch, and stop loudly on worker death - #1471
Merged
Conversation
…execute a written batch, and stop loudly on worker death Under per-table hash routing (thread.pool.size > 1, the default) the offset commit could pass rows that were still queued or in flight, and a batch that was written but not yet commit-eligible was re-inserted on every tick. Data loss (commit past unwritten rows). Commit eligibility was decided by envelope-timestamp overlap among the batches a worker had already PICKED UP. One Debezium batch becomes one routed group per table on different workers' queues, and only the last row of the whole batch carried the terminal marker. An idle worker finished its small group, saw no in-flight overlap (the sibling group was still queued on a busy worker, so it was in neither map), acknowledged the terminal row, and the offset was committed at the end of the batch; a crash before the busy worker dequeued its group lost those rows. Equal timestamps (strict >) never blocked each other either. Fix: a monotone handoff sequence assigned on the Debezium thread (binlog order) when a list is handed to the writers (registerHandoff, BEFORE any group is enqueued). A worker reports a written group once (checkIfBatchCanBeCommitted); a unit whose groups are all written is acknowledged only when it is the head of the outstanding set, otherwise parked; every acknowledgement drains parked units in sequence order. A unit is acknowledged as a whole, in binlog order, with one markBatchFinished, so staged offsets are monotone even though routed groups interleave in the binlog (Debezium 3.1.3 markBatchFinished is policy-gated and carries no per-batch state). The timestamp-overlap predicate, the pick-up registry and the handoff counter are deleted; hasUnwrittenBatches() is "the outstanding set is non-empty". A committer-bearing batch that bypassed registration is rejected (Kafka Connect batches, which have no committer, are unaffected). Duplicate rows (written-once). processBatch cleared currentBatch only when the batch was commit-eligible and both run loops re-enter processBatch while currentBatch != null, so a written-but-parked batch was re-executed -- and its rows re-inserted -- every buffer.flush.time.ms (30 ms) until it became committable (rows present 3x in non-FINAL reads, duplicate raw rows on hot tables, write amplification). The worker now drops a written batch before handing it to the FIFO; only a FAILED write keeps currentBatch for retry. Silent stall on FATAL. A FATAL classification rethrew from a scheduleAtFixedRate task; the executor cancelled the future and told nobody, the batch stayed outstanding, later units stayed parked, the worker's queue filled and the Debezium thread blocked in put. The capture now retains the workers' ScheduledFutures and checks them first thing in every handleChangeEventBatch, re-raising the worker's cause so the engine stops through its completion callback; currentBatch is no longer cleared before the rethrow. 241 MEMORY_LIMIT_EXCEEDED is reclassified retriable (it is usually concurrent memory pressure, not a property of the batch). Retry storm. Retriable failures re-ran the same batch every 30 ms. Retries of the same batch now back off exponentially: batch.retry.backoff.initial.ms (500) doubling to batch.retry.backoff.max.ms (30000), no attempt cap (offsets never pass an unwritten batch, so retrying forever is safe and the stall is visible). Spec 10.02 now describes what exists, including the head-of-line blocking consequence of per-thread FIFO queues. Routing. RoutedBatch.calculateThreadId uses Math.floorMod: a routing key whose hashCode is Integer.MIN_VALUE produced a negative index. Formal verification (Replication/OffsetFifo.lean, no sorry, axioms at most [propext, Quot.sound]): commit_never_passes_outstanding, acked_downward_closed, commitPoint_acked, outstanding_ge_commitPoint, write_at_most_once, written_batch_not_reexecuted, old_overlap_rule_unsafe (equal-timestamp counterexample to the deleted rule), fifo_acknowledges_in_handoff_order. Wired into the lakefile roots, validate_specs.py and the Lean README. Note: the lean_lib has no default target, so the proofs build with `lake build Replication`. Tests (all mutation-checked red -> green): OffsetHandoffOrderTest, ParkedBatchWrittenOnceTest, HandedOffBatchVisibilityTest, OffsetBatchIdentityTest (adapted, identity keying still covered), RetryBackoffTest, WorkerDeathIsLoudTest, RoutedBatchTest (converted to JUnit 5 -- the module has no vintage engine, so it never ran before), ClickHouseErrorClassifierTest, HashRoutingPerTableOrderingTest. Specs: 09.01 rewritten (handoff-sequence FIFO, written-once), 09.04, 01.05 (shipped defaults buffer.flush.time.ms=30 / buffer.max.records=100000; the legacy records queue is unused in routing mode; loud handoff failure), 01.06 (terminal marker per handed-off unit, precisely), 03.01 (worker liveness), 03.03, 10.01, 10.02 (rewritten), 10.04, 11.03, CONSTITUTION I8/I12, doc/configuration.md.
minguyen9988
added a commit
that referenced
this pull request
Sep 20, 2026
…d-ddl-failure Forward-merge of #1471 (handoff-sequence FIFO, written-once) into the DDL barrier branch. Conflicts resolved by keeping both sides: lakefile roots, validate_specs.py LEAN_MODULES, Lean README tree and theorem table, CONSTITUTION module list, spec 01.06 verification list. DdlDrainDeadlockTest adapted to the merged offset API: RoutedBatch now carries its handoff sequence, and the unacknowledged-batch drain test registers a real unit with registerHandoff() and releases it through checkIfBatchCanBeCommitted() instead of the removed handoff counter; the FIFO statics are reset reflectively so no earlier test can leak a unit. Verified on the merged tree: lake build (13 modules, zero sorry), validate_specs.py PASSED, and the 15 regression classes of both PRs green (52 tests).
minguyen9988
added a commit
that referenced
this pull request
Sep 20, 2026
Forward-merge of #1470 and #1471 into the translator branch. The three conflicts were the shared registration points, resolved by keeping every side: lakefile roots (DdlBarrier, OffsetFifo, DdlTranslation), validate_specs.py LEAN_MODULES, and the Lean README module tree. Verified on the merged tree: lake build (14 modules, zero sorry), validate_specs.py PASSED, MySqlDDLParserListenerImplTest 135/135 (1 skipped), DdlReplayIdempotencyTest 9/9, and the merged regression classes of #1470/#1471 (DdlDrainDeadlockTest 6, DdlFailureLoudTest 4, ControlRecordLogLevelTest 3, ParkedBatchWrittenOnceTest 1, OffsetHandoffOrderTest 6, HashRoutingPerTableOrderingTest 4, WorkerDeathIsLoudTest 2) all green.
minguyen9988
added a commit
that referenced
this pull request
Sep 20, 2026
…-drift Forward-merge of #1470, #1471 and #1472 into the governance branch. Conflicts resolved by keeping both sides where both are true: CONSTITUTION section 5 (module bullets for DdlBarrier, OffsetFifo and DdlTranslation plus the coverage table, whose I5 and I8 rows are now Lean-covered), specs 03.01, 03.05, 10.04 and 11.03 (corrected Codebase Mapping and real test citations from both sides; theorem list renumbered). validate_specs.py keeps the rewritten validator; the hardcoded LEAN_MODULES list is obsolete because the rewrite discovers modules from the lakefile and the directory. Two consequences of the stricter validator on the merged tree are fixed here: spec 01.06 cited ControlRecordLogLevelTest.isControlRecordClassification, which the class does not declare (replaced by an honest gap line), and eight allowlist entries for specs that have since landed are pruned. Versioning specs 01.04, 02.01, 02.02, 02.04 and 02.06 now state the governing compatibility constraint: the emitted _version domain (ts_ms * 1_000_000 + counter with the 2.8.0 seeds) is the contract shared with 2.8.0, 2.9.1 and 2.10.x and must not change in either direction; the restart carry is inherited 2.8.0 behaviour preserved by design, and any improvement is confined to where the restart floor starts and must ship with an upgrade/downgrade matrix. Verified on the merged tree: validate_specs.py PASSED (3 allowlisted warnings for the still-open writer PR), validator self-tests 39 passed, lake build 14 modules zero sorry, DebeziumChangeEventCaptureTest 23/23, SequenceSeedOverflowTest 3/3, and merged regression classes green (MySqlDDLParserListenerImplTest 135, DdlDrainDeadlockTest 6, ControlRecordLogLevelTest 3, OffsetHandoffOrderTest 6). The 11 destructive-looking sites in this merge are all inherited from the already-merged #1472: DDL-translator unit-test expectation strings (DROP TABLE IF EXISTS / DROP COLUMN IF EXISTS / DROP INDEX inside ALTER fixtures) and the Lean inductive constructor tag dropColumn in DdlTranslation.lean. None executes anything; no runtime drop path is added by this merge. Destructive-Op-Check: sites=11; result=pass
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Defects fixed (spec first; every fix mutation-checked: revert = red, restore = green)
Verification