Repository navigation
Conversation
When disk protection rejects a write with DISK_WRITE_LOCKED, the Log/KV
writers apply a bounded exponential backoff before retrying that bucket,
instead of hot-looping and hammering the rejecting server.
Write-throttling is unified in a new WriteThrottleController that owns two
independent gates per bucket:
- KV backpressure (wall-clock, quadratic) for KV upsert/delete
- disk-write backoff (monotonic nanos, exponential, never-shorten,
ceil-to-ms) for DISK_WRITE_LOCKED rejections
Per bucket the effective delay is max(kvRemainingMs, diskRemainingMs); the
accumulator wakes on the min across buckets. bucketReady keeps leader-first
ordering: a leaderless bucket is still reported to unknownLeaderTables while
a gate is pending, so metadata refresh (which sends no data) stays orthogonal
to throttling.
Adds ConfigOptions for the backoff bounds, documentation, unit coverage in
RecordAccumulatorTest/SenderTest, and DiskWriteBackoffITCase.
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Per-batch metadata invalidation repeatedly rebuilds cluster metadata during multi-bucket failures.
Review effort: Balanced
Findings: 1
Open (1)
What changed in this PR
Adds per-bucket retry backoff to Java Log and KV writers, preventing retry storms during disk protection and transient failures.
Changes:
- Adds configurable exponential retry and disk-protection backoff.
- Unifies KV pressure and retry gates in
WriteThrottleController. - Adds unit and integration coverage for retry timing, recovery, and routing.
| File | Description |
|---|---|
website/docs/maintenance/configuration.md |
Documents writer backoff options. |
fluss-common/.../ConfigOptions.java |
Defines four client configuration options. |
fluss-client/.../WriteThrottleController.java |
Implements unified per-bucket throttling. |
fluss-client/.../RecordAccumulator.java |
Applies gates during readiness and draining. |
fluss-client/.../Sender.java |
Installs backoff and re-enqueues failed writes. |
fluss-client/.../RecordAccumulatorTest.java |
Tests gate timing and configuration. |
fluss-client/.../SenderTest.java |
Tests retry behavior and wakeups. |
fluss-client/.../DiskWriteBackoffITCase.java |
Tests real RPC rejection and recovery. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| metadataUpdater.invalidPhysicalTableBucketMeta( | ||
| Collections.singleton(writeTargetPath)); |
|
@ coding-otter-bot review |
|
@coding-otter-bot review |
coding-otter-bot
left a comment
coding-otter-bot
left a comment
There was a problem hiding this comment.
Automated code review
Problem
Java Log and KV writers retried rejected writes immediately, hot-looping and re-sending full payloads on every sender cycle for every queued batch on a bucket while a TabletServer was in disk write protection or another transient write failure.
Approach
- Added
WriteThrottleController, owning three per-TableBucketgates: KV backpressure (wall-clock, latest-wins, quadratic), a general retriable-write backoff, and a disk-write backoff on monotonic nanos;remainingDelayMs()/isGated()is the single read path consumed by bothready()anddrain(). RecordAccumulatorvalidates and builds the twoExponentialBackoffinstances, delegates gate install/read to the controller, and keeps the earliest effective per-bucket deadline (with thebatchTimeoutMscap so zero batch timeout cannot busy-loop).Senderinstalls the backoff before re-enqueueing (so the exponent uses the pre-retry attempt count), invalidates the actual RPC target metadata before publishing the retry, and wakes the sender after re-enqueue and after a KV pressure update that shortens the effective gate.- Four new Java client config options plus documentation for retry and disk-protection backoff.
- Coverage added in
RecordAccumulatorTest(gate math/independence/eviction),SenderTest(response-error and RPC-failure disk rejection, idempotent/non-idempotent, exhaustion, historical targets, wakeups) andDiskWriteBackoffITCase(real-RPC recovery for log/KV).
Review coverage
Limited: no additional finding is implied, but some material paths remain unresolved.
- Audit omitted candidates without evidence-backed rejection reasons: close-while-locked-assertion-too-loose.
Findings
1 inline finding survived the final audit.
| // actual RPC target first so the retry cannot race ahead using stale metadata. A | ||
| // historical batch remains keyed by its original partition path in the | ||
| // accumulator, while its RPC is sent to the internal historical partition. | ||
| metadataUpdater.invalidPhysicalTableBucketMeta( |
There was a problem hiding this comment.
MEDIUM: Retryable invalid-metadata write responses now rebuild the Cluster snapshot once per failing bucket (up to once per bucket in the response) instead of once per response.
handleProduceLogResponse/handlePutKvResponse iterate every bucket response and call handleWriteBatchException per bucket, and the changed code invalidates the RPC target inline via metadataUpdater.invalidPhysicalTableBucketMeta(Collections.singleton(writeTargetPath)) instead of accumulating writeTargetPath into invalidMetadataTablesSet for the single post-loop call; the set that handleWriteBatchException now returns is empty on this path, so the trailing aggregated invalidate is a no-op. Each invalidate rebuilds a Cluster by copying availableLocationsByPath (new HashMap plus a defensive ArrayList copy per retained path), bucketCountByTableOrPartition, tableIdByPath, partitionsIdByPath and aliveTabletServersById, so an N-bucket invalid-metadata response costs N O(cluster-metadata) rebuilds where the base cost one. The invalidate-before-retry ordering is required, but per-bucket fan-out is not: the union of paths could still be invalidated once before the batches are re-enqueued.
Failure scenario: A ProduceLog/PutKv response to a node rejects M buckets with a retryable InvalidMetadataException (e.g. NOT_LEADER_OR_FOLLOWER after a leader change). Precondition: the cluster snapshot holds B bucket locations. Operation: the per-bucket loop calls handleWriteBatchException M times, each calling invalidPhysicalTableBucketMeta(Collections.singleton(path)); every call rebuilds a cluster of O(B) entries. Actual result: M full Cluster rebuilds instead of 1, all on the sender/RPC callback thread. Violated contract: the base per-response invalidation cost contract; response handling becomes O(M*B) allocations and cluster-map copies under a metadata storm.
Suggested fix: Collect writeTargetPath into a per-response Set inside handleWriteBatchException (or return it as before) and perform one metadataUpdater.invalidPhysicalTableBucketMeta(set) after the response loop and before the batches are re-enqueued, preserving the invalidate-before-retry ordering while restoring one Cluster rebuild per response.
Confidence: 90/100
Automated code review could not completeThe review for commit Failure code: After the problem is resolved, post a new comment to retry:
|
Automated code review could not completeThe review for commit Failure code: After the problem is resolved, post a new comment to retry:
|
Purpose
Linked issue: #4278
This PR reproduces #4293 at commit
ac91ce5b146bba074cbf6c706867c28156d82937without code changes. All four original commits by @fhan688 are preserved.Java Log and KV writers currently retry rejected writes immediately, repeatedly resending payloads during disk protection or other transient write failures. This change adds bounded retry delays per physical table bucket so writes can recover without a tight retry loop.
Brief change log
WriteThrottleController; both readiness and batch draining respect the active delays.Tests
RecordAccumulatorTest(29),SenderTest(68), andDiskWriteBackoffITCase(3), with no failures or errors.API and Format
Adds four Java client configuration options:
client.writer.retry-backoff,client.writer.retry-backoff-max,client.writer.disk-write-locked.backoff, andclient.writer.disk-write-locked.backoff-max. No protocol or storage format changes; existing error codes are reused.Documentation
Documents the retry and disk-protection options in
website/docs/maintenance/configuration.md.