{{ message }}
Repository navigation
Conversation
thiskian
requested review from
1sonofqiu,
Gezi-lzq and
superhx
as code owners
October 20, 2026 14:14
## Why Snapshot replicas are updated outside the ordinary partition-stop path. Removing or replacing one can leave an existing delayed Fetch waiting until its long-poll timeout, although the old replica can no longer serve it. ## What Recheck delayed Fetch and Produce requests when a snapshot replica changes identity. An unchanged replica does not trigger completion. ## How Publish the new snapshot-replica mapping before rechecking delayed requests. Run the recheck outside the map's compute callback to avoid introducing a map/purgatory lock-order inversion. Existing request eligibility and error handling decide whether the request completes. ## Testing Regression coverage exercises removal, replacement with a fenced replica, publication before completion, and an unchanged mapping that must leave the Fetch waiting. Local verification passed: both snapshot-transition regression tests, Checkstyle and Java formatting. The repository's `rat` task was invoked and skipped by upstream configuration. ```bash ./gradlew --build-cache core:S3UnitTest --tests 'unit.kafka.server.streamaspect.ElasticReplicaManagerTest.testSnapshot*' ./gradlew --build-cache rat checkstyleMain checkstyleTest spotlessJavaCheck ``` ## Reviewer notes Start with `ElasticReplicaManager.computeSnapshotReadPartition`, then the snapshot-transition tests in `ElasticReplicaManagerTest`. This change has no prerequisite in this series. Related to AutoMQ#3669. AI assistance: OpenAI Codex (GPT-6) was used to prepare the contribution split, adapt the changes to main, and update the tests and descriptions.
## Why A partition can begin shutdown while an accepted append is still registering its delayed Produce. Failed Produce responses can also reach clients before a usable replacement-leader hint is published. ## What Coordinate accepted Produce registration with partition shutdown, drain all pending durable callbacks before unhosting, and briefly retain failed responses while waiting for a published leader hint. ## How Main already waits for the final append acknowledgement outside the log lock. Retain that close ordering, its acknowledgement chain and truncate-epoch guards; add the pre-unhost barrier and tracking for every pending callback. A partition-owned registration barrier stops new admission and waits for reservations already accepted by the enclosing append. Drain all pending durable callbacks and the acknowledgement executor before unhosting, retaining the ordinary high-watermark and ISR checks. Timeout and failure retain the normal unhost/close path. Leader-hint waits apply to failed ZeroZone V2 Produce responses that support hints. Waits are bounded by the request budget and a one-second ceiling, with at most 64 admitted responses. They preserve the original error, release capacity on cancellation or failure, and track actual socket-send completion during shutdown. A draining local partition is not advertised as a usable retry destination. ## Testing Coverage includes late registration, callback failures, concurrent accepted writes, lock availability during drain, out-of-order persistence completion, bounded hint admission and expiry, socket-send completion, and response error preservation. Local verification passed: the focused registration, drain, log and API suites, both tagged partition-drain regressions, Checkstyle and Java formatting. The repository's `rat` task was invoked and skipped by upstream configuration. ```bash ./gradlew --build-cache core:S3UnitTest \ --tests 'kafka.automq.zerozone.AcceptedProduceDrainTest' \ --tests 'kafka.automq.zerozone.LeaderHintRecoveryTest' \ --tests 'kafka.log.streamaspect.ElasticLogTest' \ --tests 'kafka.log.streamaspect.ElasticUnifiedLogTest' \ --tests 'kafka.cluster.PartitionTest.testAcceptedProduceDrain*' \ --tests 'unit.kafka.server.streamaspect.ElasticKafkaApisTest' \ --tests 'unit.kafka.server.streamaspect.ElasticReplicaManagerTest' ./gradlew --build-cache rat checkstyleMain checkstyleTest spotlessJavaCheck ``` ## Reviewer notes Depends on AutoMQ#3670. This PR targets `main`. Its cumulative diff includes the preceding snapshot-fetch correction. The individual commit isolates this contribution. Merge the stack in order, then rebase the remaining branches onto `main` and repeat verification. Read `AcceptedProduceDrain`, the partition/replica-manager registration path, and the `ElasticLog`/`ElasticUnifiedLog` drain path first. Then review `LeaderHintRecovery`, `ElasticKafkaApis`, and the shutdown ordering in `BrokerServer`. WAL acceleration is covered by a subsequent PR. Related to AutoMQ#3669. AI assistance: OpenAI Codex (GPT-6) was used to prepare the contribution split, adapt the changes to main, and update the tests and descriptions.
thiskian
added a commit
to thiskian/automq
that referenced
this pull request
Oct 21, 2026
## Why Optional batching can add waits at multiple stages of a partition handoff: router batching, physical object WALs, serial metadata writes, and snapshot cache-pressure commits. Recovery priority can disappear when records enter another batching layer. ## What Carry recovery scheduling hints through both router and data WAL stages, request prompt durable metadata appends, and coalesce snapshot cache-pressure commits. ## How Keep urgent metadata appends distinct from coalesced recovery appends. Carry their scheduling hints through dedicated append contexts and the router frame, retaining ordinary buffer ownership, upload admission, durability and acknowledgement semantics. Manual WALs remain under router-owned flush control. A broker-local recovery policy caps batching at 50 ms for at most three two-thirds-second bursts per rolling minute. Handoffs, proxy topology changes and controlled drain share that bounded policy. Pending batches can be shortened, and faster configured intervals are preserved. Coalesce snapshot cache-pressure commits across sessions with one running and one pending batch and at most one start per second. Requests arriving after upload starts join the next batch. Group close-time uploads and allow the first adaptive upload after storage startup to avoid pacing itself at its accumulation rate. Keep main's asynchronous stream-open checks, fast metadata handoff, admitted-mutation freeze and offset-aware cache publication. Only metadata append scheduling and compaction urgency change in that lifecycle. ## Testing Coverage exercises priority propagation through router and physical WALs, pending-batch shortening, manual flush ownership, durable completion, upload failures, recovery budgets, metadata compaction and close, snapshot commit coalescing, and admitted-mutation close ordering. Local verification passed: `metadata:S3UnitTest`, `core:S3UnitTest`, `s3stream:test`, Checkstyle, Java formatting and storage SpotBugs. The repository's `rat` task was invoked and skipped by upstream configuration. ```bash ./gradlew --build-cache metadata:S3UnitTest core:S3UnitTest s3stream:test ./gradlew --build-cache rat checkstyleMain checkstyleTest spotlessJavaCheck s3stream:spotbugsMain ``` ## Reviewer notes Depends on AutoMQ#3671. The stack begins with AutoMQ#3670. This PR targets `main`. Its cumulative diff includes the preceding snapshot-fetch and shutdown-coordination corrections. The individual commit isolates this contribution. Merge the stack in order, then rebase the remaining branches onto `main` and repeat verification. Suggested reading order: 1. `AppendContext`, `WriteAheadLog`, `Writer`, and `RecoveryBatchingWindow` establish the scheduling contract. 2. `DefaultWriter`, `ObjectRouterChannel`, and `MultiBucketsRouterChannel` preserve it through batching and durable completion. 3. `RecoveryProduceScheduling`, `LeaderRecoveryBatching`, and the ZeroZone routing hooks trigger bounded recovery work. 4. `MetaStream`, `SnapshotReadCommitter` and `S3Storage` remove the remaining scheduling waits while retaining main's handoff lifecycle. Shorter batching can increase object-store PUT frequency. The issue's latency measurements cover the combined 1.7 implementation. They do not measure this main port or this PR's individual contribution. Related to AutoMQ#3669. AI assistance: OpenAI Codex (GPT-6) was used to prepare the contribution split, adapt the changes to main, and update the tests and descriptions.
thiskian
force-pushed
the
kabdolahi/3669-produce-handoff-drain
branch
from
October 21, 2026 10:18
b6bf0df to
66d80ce
Compare
This branch has not been deployed
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.
Why
A partition can begin shutdown while an accepted append is still registering its delayed Produce. Failed Produce responses can also reach clients before a usable replacement-leader hint is published.
What
Coordinate accepted Produce registration with partition shutdown, drain all pending durable callbacks before unhosting, and briefly retain failed responses while waiting for a published leader hint.
How
Main already waits for the final append acknowledgement outside the log lock. Retain that close ordering, its acknowledgement chain and truncate-epoch guards; add the pre-unhost barrier and tracking for every pending callback.
A partition-owned registration barrier stops new admission and waits for reservations already accepted by the enclosing append. Drain all pending durable callbacks and the acknowledgement executor before unhosting, retaining the ordinary high-watermark and ISR checks. Timeout and failure retain the normal unhost/close path.
Leader-hint waits apply to failed ZeroZone V2 Produce responses that support hints. Waits are bounded by the request budget and a one-second ceiling, with at most 64 admitted responses. They preserve the original error, release capacity on cancellation or failure, and track actual socket-send completion during shutdown. A draining local partition is not advertised as a usable retry destination.
Testing
Coverage includes late registration, callback failures, concurrent accepted writes, lock availability during drain, out-of-order persistence completion, bounded hint admission and expiry, socket-send completion, and response error preservation.
Local verification passed: the focused registration, drain, log and API suites, both tagged partition-drain regressions, Checkstyle and Java formatting. The repository's
rattask was invoked and skipped by upstream configuration.Reviewer notes
Incremental review: this contribution only.
Depends on #3670.
This PR targets
main. Its cumulative diff includes the preceding snapshot-fetch correction. The individual commit isolates this contribution. Merge the stack in order, then rebase the remaining branches ontomainand repeat verification.Read
AcceptedProduceDrain, the partition/replica-manager registration path, and theElasticLog/ElasticUnifiedLogdrain path first. Then reviewLeaderHintRecovery,ElasticKafkaApis, and the shutdown ordering inBrokerServer. WAL acceleration is covered by a subsequent PR.Related to #3669.
AI assistance: OpenAI Codex (GPT-6) was used to prepare the contribution split, adapt the changes to main, and update the tests and descriptions.