Repository navigation
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR adds an additional cold-path limiter for Inkless fetches: a byte-rate (bandwidth) limiter that meters the physical object-storage read size, complementing the existing request-rate limiter (QPS/cost). It also wires this new limiter into both consumer lagging fetches and diskless consolidation cold-path fetches, with supporting metrics, documentation, and tests.
Changes:
- Add a configurable byte-rate limiter for lagging/cold-path fetches and plumb it through Reader → FetchPlanner (including “oversized fetch” handling).
- Introduce new metrics to attribute throttling to request-rate vs byte-rate limiters while keeping a combined wait-time histogram.
- Add new configuration keys (Inkless + server-side consolidation) and update docs/tests accordingly.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java | Updates Reader constructor call sites for the new byte-rate limiter parameter. |
| storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java | Adds byte-rate limiter test coverage and updates FetchPlanner construction call sites. |
| storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java | Validates default/explicit values for the new lagging consumer byte-rate config. |
| storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java | Builds and passes the new byte-rate bucket (and capacity) into FetchPlanner. |
| storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java | Adds throttle/oversized meters and updates wait-time metric description. |
| storage/inkless/src/main/java/io/aiven/inkless/consume/FetchPlanner.java | Applies request-rate and byte-rate limiting (tryConsume-first, then blocking consume) and records per-limiter throttle/oversized metrics. |
| storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java | Wires the new InklessConfig byte-rate limit into Reader construction. |
| storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java | Adds config definition + startup WARN heuristic if byte-rate limit is below produced object size. |
| server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java | Adds the consolidation cold-path byte-rate config key and definition. |
| docs/inkless/metrics.rst | Documents the new InklessFetch byte-rate/request-rate throttling metrics. |
| docs/inkless/configs.rst | Documents fetch.lagging.consumer.byte.rate.limit. |
| core/src/main/scala/kafka/server/ReplicaManager.scala | Passes consolidation byte-rate limit through to Reader for consolidation fetches. |
| core/src/main/scala/kafka/server/KafkaConfig.scala | Exposes the new consolidation byte-rate limit config via KafkaConfig. |
Comment on lines
+1806
to
+1815
| // Large capacity with a fetch cost near it: after draining, the fetch's tryConsume | ||
| // fails (only a tiny fraction refills during scheduling), so it must block on refill. | ||
| final long capacity = 1_000_000L; | ||
| final long fetchByteSize = 500_000L; | ||
| final Bucket byteRateLimiter = Bucket.builder() | ||
| .addLimit(limit -> limit.capacity(capacity).refillGreedy(capacity, Duration.ofSeconds(1))) | ||
| .build(); | ||
| // Pre-drain: consume the full bucket so the next consume must wait for a refill. | ||
| assertThat(byteRateLimiter.tryConsume(capacity)).isTrue(); | ||
|
|
Comment on lines
+1838
to
+1847
| // Low rate so a single token takes ~500ms to refill: after draining, the fetch's | ||
| // tryConsume(1) fails during scheduling and must block on refill. The 500ms margin | ||
| // keeps the test robust even if the lagging executor is slow to pick up the task. | ||
| final int capacity = 2; | ||
| final Bucket rateLimiter = Bucket.builder() | ||
| .addLimit(limit -> limit.capacity(capacity).refillGreedy(capacity, Duration.ofSeconds(1))) | ||
| .build(); | ||
| // Pre-drain the request bucket. | ||
| assertThat(rateLimiter.tryConsume(capacity)).isTrue(); | ||
|
|
…cold-path fetches The lagging cold path only had a request-rate limit (storage GET cost/QPS), leaving nothing to bound object-storage read bandwidth per node. Add an independent byte-rate limiter that meters the physical S3 GET range size (byteRange().size(), including bounding-range read amplification). - Consumer: new fetch.lagging.consumer.byte.rate.limit (default 0 = disabled). - Consolidation: new diskless.consolidation.fetch.lagging.byte.rate.limit, distinct from the existing rate.limit.bytes.per.second quota which meters processed record bytes, not physical reads. - Second Bucket4j bucket (capacity = byteRate); acquire is tryConsume-first so each limiter records a per-limiter throttle hit while sharing one wait-time histogram. Oversized fetches (byteRange > capacity) are charged a full bucket and let through instead of deadlocking, tracked via an oversized meter. - Startup WARN when the consumer byte-rate limit is below produce.buffer.max.bytes. - Hedges remain exempt from both limiters.
jeqo
force-pushed
the
jeqo/lagging-byte-rate-limit
branch
from
July 27, 2026 08:46
7377bd3 to
5ca6497
Compare
|
This PR is being marked as stale since it has not had any activity in 90 days. If you If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact). If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed. |
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.
The lagging cold path only had a request-rate limit (storage GET cost/QPS), leaving nothing to bound object-storage read bandwidth per node. Add an independent byte-rate limiter that meters the physical S3 GET range size (byteRange().size(), including bounding-range read amplification).