Skip to content

[client] Back off writes rejected by disk protection #4533

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
luoyuxia wants to merge 4 commits into from
Closed

[client] Back off writes rejected by disk protection #4533

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a0f8439
[client] Back off writes rejected by disk protection
fhan688 Sep 16, 2026
99e067b
fix(client): invalidate metadata before re-enqueuing write retries
fhan688 Sep 16, 2026
f5f3ee7
fix(client): wake sender when KV pressure shortens throttle deadline
fhan688 Sep 29, 2026
ac91ce5
fix(client): back off retriable writes per bucket
fhan688 Sep 29, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
291 changes: 180 additions & 111 deletions fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java
View file
Open in desktop

Large diffs are not rendered by default.

80 changes: 55 additions & 25 deletions fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import javax.annotation.concurrent.GuardedBy;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand Down Expand Up @@ -247,6 +248,7 @@ private void sendWriteData() throws Exception {
// Refresh per-bucket throttle entries against the current cluster snapshot,
// dropping any whose bucket has disappeared from metadata.
accumulator.maybeEvictStaleThrottles(clusterSnapshot);
accumulator.maybeEvictExpiredWriteBackoffs();

// get the list of buckets with data ready to send.
ReadyCheckResult readyCheckResult = accumulator.ready(clusterSnapshot);
Expand Down Expand Up @@ -359,6 +361,7 @@ private void reEnqueueBatch(ReadyWriteBatch readyWriteBatch) {
boolean reEnqueued = accumulator.reEnqueue(readyWriteBatch);
maybeRemoveFromInflightBatches(readyWriteBatch);

wakeup();
if (reEnqueued) {
// metrics for retry record count.
writerMetricGroup
Expand Down Expand Up @@ -629,6 +632,7 @@ private void handlePutKvResponse(
long tableId,
Map<WriteBatchKey, ReadyWriteBatch> writeBatchesByKey) {
Set<PhysicalTablePath> invalidMetadataTablesSet = new HashSet<>();
boolean eligibilityAdvanced = false;
for (PbPutKvRespForBucket respForBucket : putKvResponse.getBucketsRespsList()) {
TableBucket tb =
new TableBucket(
Expand All @@ -638,7 +642,7 @@ private void handlePutKvResponse(

// Update backpressure throttle from pressure signal
if (respForBucket.hasPressure()) {
accumulator.updateThrottle(tb, respForBucket.getPressure());
eligibilityAdvanced |= accumulator.updateThrottle(tb, respForBucket.getPressure());
}

ReadyWriteBatch writeBatch =
Expand All @@ -664,6 +668,12 @@ private void handlePutKvResponse(
}
}
metadataUpdater.invalidPhysicalTableBucketMeta(invalidMetadataTablesSet);
// A fresher latest-wins KV pressure signal may shorten the effective gate. Wake only after
// the whole response is applied so the sender observes completed batches and invalidated
// metadata together with the new deadline.
if (eligibilityAdvanced) {
wakeup();
}
}

private void handleWriteRequestException(Throwable t, List<ReadyWriteBatch> writeBatches) {
Expand Down Expand Up @@ -746,13 +756,27 @@ private Set<PhysicalTablePath> handleWriteBatchException(
} else if (canRetry(readyWriteBatch, error.error())) {
// if batch failed because of retrievable exception, we need to retry send all those
// batches.
LOG.warn(
"Get error write response on table bucket {}, retrying ({} attempts left). Error: {}",
readyWriteBatch.tableBucket(),
retries - writeBatch.attempts(),
error.formatErrMsg());

if (error.exception() instanceof InvalidMetadataException) {
if (error.exception() instanceof UnknownTableOrBucketException) {
LOG.warn(
"Received unknown table or bucket error in write request on bucket {}. The table-bucket may not exist.",
readyWriteBatch.tableBucket());
} else {
LOG.warn(
"Received invalid metadata error in write request on bucket {}. "
+ "Going to request metadata update.",
readyWriteBatch.tableBucket(),
error.exception());
}
// Re-enqueuing publishes the retry to the sender and wakes it up. Invalidate the
// 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(

@coding-otter-bot coding-otter-bot Oct 10, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Collections.singleton(writeTargetPath));
Comment on lines +775 to +776
}
if (!idempotenceManager.idempotenceEnabled()) {
prepareWriteRetry(readyWriteBatch, error);
reEnqueueBatch(readyWriteBatch);
} else if (idempotenceManager.hasWriterId(writeBatch.writerId())) {
// If idempotence is enabled only retry the request if the current writer id is
Expand All @@ -761,6 +785,7 @@ private Set<PhysicalTablePath> handleWriteBatchException(
"Retrying batch to table-bucket {}, Batch sequence : {}",
readyWriteBatch.tableBucket(),
writeBatch.batchSequence());
prepareWriteRetry(readyWriteBatch, error);
reEnqueueBatch(readyWriteBatch);
} else {
Exception exception =
Expand All @@ -771,24 +796,6 @@ private Set<PhysicalTablePath> handleWriteBatchException(
writeBatch.writerId(), idempotenceManager.writerId()));
failBatch(readyWriteBatch, exception, false);
}

if (error.exception() instanceof InvalidMetadataException) {
if (error.exception() instanceof UnknownTableOrBucketException) {
LOG.warn(
"Received unknown table or bucket error in write request on bucket {}. The table-bucket may not exist.",
readyWriteBatch.tableBucket());
} else {
LOG.warn(
"Received invalid metadata error in write request on bucket {}. "
+ "Going to request metadata update.",
readyWriteBatch.tableBucket(),
error.exception());
}
// A historical batch remains keyed by its original partition path in the
// accumulator, but its RPC is sent to the internal historical partition. Invalidate
// the actual RPC target so the retry refreshes the historical bucket metadata.
invalidMetadataTables.add(writeTargetPath);
}
} else {
LOG.warn(
"Get error write response on table bucket {}, fail. Error: {}",
Expand All @@ -802,6 +809,29 @@ private Set<PhysicalTablePath> handleWriteBatchException(
return invalidMetadataTables;
}

private void prepareWriteRetry(ReadyWriteBatch batch, ApiError error) {
long retryBackoffMs = accumulator.backoffAfterRetriableWrite(batch);
if (error.error() == Errors.DISK_WRITE_LOCKED) {
long diskBackoffMs = accumulator.backoffAfterDiskWriteLocked(batch);
LOG.warn(
"Get error write response on table bucket {}, disk backoff {} ms and "
+ "retry backoff {} ms ({} attempts left). Error: {}",
batch.tableBucket(),
diskBackoffMs,
retryBackoffMs,
retries - batch.writeBatch().attempts(),
error.formatErrMsg());
} else {
LOG.warn(
"Get error write response on table bucket {}, retrying after {} ms "
+ "({} attempts left). Error: {}",
batch.tableBucket(),
retryBackoffMs,
retries - batch.writeBatch().attempts(),
error.formatErrMsg());
}
}

/**
* Rechecks unknown-leader partitions after a bulk metadata update reports {@link
* PartitionNotExistException}, and handles missing partitions for tables with historical
Expand Down
Loading
Loading