[server][client] Add cooperative KV backpressure based on RocksDB L0

[platinumhamburg]

Introduce a two-tier backpressure model to keep RocksDB write stall from blocking RPC handler threads:

• Tier 1 (piggyback): server reports a normalized [0, 1) pressure on PbPutKvRespForBucket once L0 reaches kv.backpressure.l0-slowdown-trigger (default 8); client throttles per bucket with max_throttle * p^2 via client.writer.kv-backpressure.max-throttle (default 3s).
• Tier 2 (hard rejection): once L0 reaches RocksDB's level0_slowdown_writes_trigger, server throws StorageBackpressureException (Errors code 71, RetriableException); client applies the same max-throttle window as backoff before retry.

Add maintenance configuration and observability docs for the new server/client options and metrics.

Purpose

Linked issue: close #3454 3454

Brief change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This PR introduces cooperative KV backpressure for primary-key tables backed by RocksDB, aiming to prevent RocksDB write stalls from tying up RPC handler threads by (1) piggybacking a per-bucket normalized pressure signal on PutKv responses and (2) fast-failing writes with a new retriable StorageBackpressureException when the storage engine reaches a hard threshold. It also refactors KV flush into a shared asynchronous scheduler so KV materialization can happen off RPC threads, and adds table-level metrics plus documentation.

Changes:

• Add per-bucket KV pressure sampling (based on RocksDB L0) and propagate it via RPC (proto + server response building) and client-side throttling/backoff.
• Introduce async KV flush scheduling (KvFlushScheduler) and adjust delayed-write completion semantics to optionally wait for local KV flush (not only high watermark).
• Add metrics + tests across server/client, and document new configuration and observability signals.

Reviewed changes

Copilot reviewed 36 out of 36 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
website/docs/maintenance/observability/monitor-metrics.md	Documents new KV backpressure metrics exposed at table scope.
website/docs/maintenance/configuration.md	Documents new backpressure + async flush related configuration options.
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/FlussSparkTestBase.scala	Makes Spark UT log polling resilient to staged HW advancement with async flush.
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java	Adds test ensuring PutKv acks=1 waits for local flush (not HW).
fluss-server/src/test/java/org/apache/fluss/server/replica/KvReplicaRestoreITCase.java	Adjusts restore IT to assert PutKv bucket success and avoid backpressure gate interference.
fluss-server/src/test/java/org/apache/fluss/server/replica/KvRecoverFromRemoteLogITCase.java	Adjusts remote-log recovery IT to assert PutKv bucket success and avoid backpressure gate interference.
fluss-server/src/test/java/org/apache/fluss/server/replica/delay/DelayedWriteTest.java	Updates delayed-write metadata construction for new completion mode.
fluss-server/src/test/java/org/apache/fluss/server/metrics/group/TableMetricGroupTest.java	Adds unit test for newly exposed KV backpressure metrics surface.
fluss-server/src/test/java/org/apache/fluss/server/kv/snapshot/KvSnapshotDataUploaderTest.java	Adds regression test for uploading empty snapshot files.
fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvTest.java	Adds detailed tests for pressure curve, flush budget gate, and close-safety.
fluss-server/src/test/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBufferTest.java	Updates tests to exercise two-phase async flush prepare/complete and byte accounting.
fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java	Adds extensive tests for flush scheduling, retry/backoff, and admission gate behavior.
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java	Piggybacks pressure into PutKv response only when pressure is non-zero.
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java	Treats StorageBackpressureException as expected; counts rejections; adds delayed-write completion mode logic and pressure attachment.
fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java	Introduces flush-complete listener; enforces KV flush before HW visibility; adds local-KV-flush completion check and pressure sampling/metrics write.
fluss-server/src/main/java/org/apache/fluss/server/replica/delay/DelayedWrite.java	Adds completion modes (HW vs local KV flush) and refreshes PutKv pressure at completion time.
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java	Adds table-level KV backpressure gauge (max pressure) and counter (rejections total).
fluss-server/src/main/java/org/apache/fluss/server/metrics/group/BucketMetricGroup.java	Stores latest per-bucket pressure sample for table-level aggregation.
fluss-server/src/main/java/org/apache/fluss/server/kv/snapshot/SnapshotLocation.java	Ensures empty snapshot outputs still produce a valid KvFileHandle (no null handle).
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBWriteBatchWrapper.java	Adds “no slowdown” write option path that converts RocksDB delayed-write rejections into StorageBackpressureException.
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvBuilder.java	Plumbs Fluss L0 proactive trigger + relevant RocksDB CF options into RocksDBKv.
fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKv.java	Implements cached L0 sampling, normalized pressure calculation, flush-budget admission gate helpers, and rejection latch behavior.
fluss-server/src/main/java/org/apache/fluss/server/kv/prewrite/KvPreWriteBuffer.java	Refactors flush into two-phase prepare/complete/abort, adds pending-byte tracking and safety checks.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java	Adds async flush state machine, shared scheduler integration, write admission gate, and no-slowdown flush execution.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java	Introduces a shared KvFlushScheduler per TabletServer and wires it into tablet creation and shutdown.
fluss-server/src/main/java/org/apache/fluss/server/kv/KvFlushScheduler.java	New shared threadpool + retry-timer scheduler for async KV flush tasks.
fluss-rpc/src/main/proto/FlussApi.proto	Adds optional pressure field to PbPutKvRespForBucket.
fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java	Adds new error code 71 mapped to StorageBackpressureException.
fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PutKvResultForBucket.java	Adds pressure field to internal PutKv bucket result for response construction.
fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java	Adds table-level KV backpressure metric name constants.
fluss-common/src/main/java/org/apache/fluss/exception/StorageBackpressureException.java	New retriable exception type used for hard write rejections due to KV pressure.
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java	Adds config options for client max throttle, server flush thread count, and L0 proactive trigger.
fluss-client/src/test/java/org/apache/fluss/client/write/SenderTest.java	Adds tests for throttle application from pressure and hard-rejection backoff.
fluss-client/src/test/java/org/apache/fluss/client/write/RecordAccumulatorTest.java	Adds throttle semantics tests and config tweaks for test stability.
fluss-client/src/main/java/org/apache/fluss/client/write/Sender.java	Applies throttle on pressure signal; applies hard backoff on StorageBackpressureException; evicts stale throttles.
fluss-client/src/main/java/org/apache/fluss/client/write/RecordAccumulator.java	Implements per-bucket throttle tracking, drain/ready skipping, and metadata-based throttle eviction.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.