From 5ca6497fa7fdb044e1f5837502fab69024994ec3 Mon Sep 17 00:00:00 2001 From: Jorge Esteban Quilcate Otoya Date: Wed, 22 Jul 2026 20:42:19 +0300 Subject: [PATCH] feat(inkless:consume): add byte-rate limit for lagging/consolidation 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. --- .../main/scala/kafka/server/KafkaConfig.scala | 1 + .../scala/kafka/server/ReplicaManager.scala | 4 + docs/inkless/configs.rst | 8 + docs/inkless/metrics.rst | 75 +++--- .../kafka/server/config/ServerConfigs.java | 10 + .../aiven/inkless/config/InklessConfig.java | 48 ++++ .../aiven/inkless/consume/FetchHandler.java | 1 + .../aiven/inkless/consume/FetchPlanner.java | 44 +++- .../inkless/consume/InklessFetchMetrics.java | 55 +++- .../java/io/aiven/inkless/consume/Reader.java | 37 ++- .../inkless/config/InklessConfigTest.java | 5 + .../inkless/consume/FetchPlannerTest.java | 243 +++++++++++++++++- .../io/aiven/inkless/consume/ReaderTest.java | 6 +- 13 files changed, 483 insertions(+), 54 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 2b63e901d06..9c104e91117 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -437,6 +437,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _]) val disklessConsolidationFindBatchesMaxPerPartition: Int = getInt(ServerConfigs.DISKLESS_CONSOLIDATION_FIND_BATCHES_MAX_PER_PARTITION_CONFIG) val disklessConsolidationFetchRateLimitBytesPerSecond: Long = getLong(ServerConfigs.DISKLESS_CONSOLIDATION_FETCH_RATE_LIMIT_BYTES_PER_SECOND_CONFIG) val disklessConsolidationFetchLaggingRequestRateLimit: Int = getInt(ServerConfigs.DISKLESS_CONSOLIDATION_FETCH_LAGGING_REQUEST_RATE_LIMIT_CONFIG) + val disklessConsolidationFetchLaggingByteRateLimit: Long = getLong(ServerConfigs.DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_CONFIG) val classicRemoteStorageForceEnabled: Boolean = getBoolean(ServerConfigs.CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_CONFIG) val classicRemoteStorageForceExcludeTopicRegexes: java.util.List[String] = getList(ServerConfigs.CLASSIC_REMOTE_STORAGE_FORCE_EXCLUDE_TOPIC_REGEXES_CONFIG) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index c46b39fd805..b93678ea0e0 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -319,6 +319,10 @@ class ReplicaManager(val config: KafkaConfig, // consumer-cached block, older data is bounding-range aligned for a cheaper cold fetch. state.config().fetchLaggingConsumerThresholdMs(), config.disklessConsolidationFetchLaggingRequestRateLimit, + // Byte-rate limit on the physical object-storage read (bounding-range GET size, incl. read + // amplification). Distinct from disklessConsolidationFetchRateLimitBytesPerSecond, which limits + // the processed record bytes appended to the log via the replication quota. + config.disklessConsolidationFetchLaggingByteRateLimit, 0, // use the consolidation data pool instead // no hedged fetch for consolidation 0L, 0L, diff --git a/docs/inkless/configs.rst b/docs/inkless/configs.rst index 289f4318662..c48cb02b07b 100644 --- a/docs/inkless/configs.rst +++ b/docs/inkless/configs.rst @@ -53,6 +53,14 @@ Under ``inkless.`` * Default: false * Importance: medium +``fetch.lagging.consumer.byte.rate.limit`` + Maximum bytes per second read from object storage for lagging consumer data fetches. This caps the storage-to-broker (ingress) throughput of the cold path to keep it below the node's baseline network bandwidth. It is independent of and complementary to fetch.lagging.consumer.request.rate.limit: the request-rate limit protects against storage GET request cost (QPS), while this byte-rate limit protects network bandwidth. Set to 0 (default) to disable byte-rate limiting. Metered by the fetched byte range, so it governs storage-to-broker ingress only (not broker-to-consumer egress). A single cold fetch covers one storage object (bounding range), so it should be set at or above the produced object size (produce.buffer.max.bytes); a lower value only throttles the stream of fetches, it cannot split an individual object below its own size. Note: hedge requests triggered by slow fetches are exempt from this limit. + + * Type: long + * Default: 0 + * Valid Values: [0,...] + * Importance: medium + ``fetch.lagging.consumer.request.rate.limit`` Maximum requests per second for lagging consumer data fetches. Set to 0 to disable rate limiting. The upper bound of 10000 req/s is a safety limit to prevent misconfiguration. For high-throughput systems, consider the relationship between this rate limit, thread pool size, and storage backend capacity. At the default rate of 200 req/s with ~50ms per request latency, this allows ~10 concurrent requests. Note: hedge requests triggered by slow fetches are exempt from this limit. In the worst case, effective storage GET rate can reach up to 2x this value. diff --git a/docs/inkless/metrics.rst b/docs/inkless/metrics.rst index 5814d024725..8c2545fe7c2 100644 --- a/docs/inkless/metrics.rst +++ b/docs/inkless/metrics.rst @@ -107,42 +107,45 @@ InklessFetch metrics io.aiven.inkless.consume:type=InklessFetchMetrics ------------------------------------------------- -=================================== ======================================================================================================================================================================================================================================================================================== -Attribute name Description -=================================== ======================================================================================================================================================================================================================================================================================== -CacheEntrySize Size of individual cache entries in bytes -CacheFetchErrorRate Rate of errors when fetching from the cache per second -CacheHitCount Rate of cache hits per second -CacheMissCount Rate of cache misses per second -CacheQueryTime Time spent querying the object cache in milliseconds -CacheSize Current number of entries in the object cache -CacheStoreTime Time spent storing entries in the object cache in milliseconds -FetchBatchesPerPartitionCount Number of batches fetched per partition -FetchCompletionTime Time spent completing the fetch response assembly in milliseconds -FetchErrorRate Rate of failed fetch requests per second -FetchFileTime Time spent fetching a file from storage in milliseconds -FetchFirstByteTime Time until the first byte is received from storage in milliseconds -FetchObjectsPerFetchCount Number of storage objects accessed per fetch request -FetchPartitionsPerFetchCount Number of partitions included in each fetch request -FetchPlanTime Time spent creating the fetch plan in milliseconds -FetchRate Rate of fetch requests processed per second -FetchResponseSize Total bytes returned to the caller in each fetch response, recorded for every fetch (0 when no data was served, e.g. a caught-up long-poll or an all-error response). -FetchTotalTime Total time spent processing a fetch request in milliseconds -FileFetchErrorRate Rate of errors when fetching files from storage per second -FindBatchesErrorRate Rate of errors when finding batches in the control plane per second -FindBatchesTime Time spent finding batch coordinates in the control plane in milliseconds -HedgeRequestRate Rate of hedged fetch requests issued per second -HedgeTotalTimeTriggeredRate Rate of hedge requests triggered by total time timeout per second -HedgeTtfbTriggeredRate Rate of hedge requests triggered by TTFB timeout per second -HedgeWonRate Rate of hedge requests that completed before the original request per second -LaggingConsumerRateLimitWaitTime Wait time for rate-limited lagging consumer requests in milliseconds -LaggingConsumerRequestRate Rate of cold-path requests (bypass the cache) per second. Under the consumer metrics group these are lagging-consumer fetches; under the consolidation group (ConsolidationFetchMetrics) these are consolidation cold fetches (cache-hit peeks are counted under RecentDataRequestRate). -LaggingConsumerRequestRejectedRate Rate of lagging consumer requests rejected due to executor unavailability per second -PartitionCorruptRecordRate Rate of partition responses returning CORRUPT_MESSAGE or INVALID_RECORD from FetchCompleter because a batch failed integrity validation (CRC / declared size, or a coalesced-run offset mismatch), with no valid prefix to serve. -PartitionPartialFetchRate Rate of partition responses that were truncated mid-batch-list due to a missing extent or validation failure on a trailing batch. Successful prefix is returned to the consumer; the trailing batches are dropped. -PartitionStorageErrorRate Rate of partition responses returning KAFKA_STORAGE_ERROR from FetchCompleter (missing batch metadata, or missing/incomplete extents so no records were constructable). Data-integrity failures are counted separately under PartitionCorruptRecordRate. -RecentDataRequestRate Rate of requests served via the hot path (recent data with cache) per second. Under the consolidation metrics group (ConsolidationFetchMetrics) this counts cache-hit peeks that reuse consumer-cached data. -=================================== ======================================================================================================================================================================================================================================================================================== +======================================== ================================================================================================================================================================================================================================================================================================================================================================= +Attribute name Description +======================================== ================================================================================================================================================================================================================================================================================================================================================================= +CacheEntrySize Size of individual cache entries in bytes +CacheFetchErrorRate Rate of errors when fetching from the cache per second +CacheHitCount Rate of cache hits per second +CacheMissCount Rate of cache misses per second +CacheQueryTime Time spent querying the object cache in milliseconds +CacheSize Current number of entries in the object cache +CacheStoreTime Time spent storing entries in the object cache in milliseconds +FetchBatchesPerPartitionCount Number of batches fetched per partition +FetchCompletionTime Time spent completing the fetch response assembly in milliseconds +FetchErrorRate Rate of failed fetch requests per second +FetchFileTime Time spent fetching a file from storage in milliseconds +FetchFirstByteTime Time until the first byte is received from storage in milliseconds +FetchObjectsPerFetchCount Number of storage objects accessed per fetch request +FetchPartitionsPerFetchCount Number of partitions included in each fetch request +FetchPlanTime Time spent creating the fetch plan in milliseconds +FetchRate Rate of fetch requests processed per second +FetchResponseSize Total bytes returned to the caller in each fetch response, recorded for every fetch (0 when no data was served, e.g. a caught-up long-poll or an all-error response). +FetchTotalTime Total time spent processing a fetch request in milliseconds +FileFetchErrorRate Rate of errors when fetching files from storage per second +FindBatchesErrorRate Rate of errors when finding batches in the control plane per second +FindBatchesTime Time spent finding batch coordinates in the control plane in milliseconds +HedgeRequestRate Rate of hedged fetch requests issued per second +HedgeTotalTimeTriggeredRate Rate of hedge requests triggered by total time timeout per second +HedgeTtfbTriggeredRate Rate of hedge requests triggered by TTFB timeout per second +HedgeWonRate Rate of hedge requests that completed before the original request per second +LaggingConsumerByteRateOversizedRate Rate of lagging consumer fetches whose byte range exceeds the byte-rate limiter capacity (fetch.lagging.consumer.byte.rate.limit) per second. Such a fetch cannot be split below its own size, so it is charged a full bucket and allowed through. A non-zero rate indicates the byte-rate limit is set below the produced object size (likely misconfiguration). +LaggingConsumerByteRateThrottledRate Rate of lagging consumer requests that had to wait on the byte-rate limiter (bandwidth protection) per second. Together with LaggingConsumerRequestRateThrottledRate, attributes which limiter is the binding constraint (the combined wait is in LaggingConsumerRateLimitWaitTime). +LaggingConsumerRateLimitWaitTime Combined wait time (request-rate plus byte-rate limiters) for rate-limited lagging consumer requests in milliseconds +LaggingConsumerRequestRate Rate of cold-path requests (bypass the cache) per second. Under the consumer metrics group these are lagging-consumer fetches; under the consolidation group (ConsolidationFetchMetrics) these are consolidation cold fetches (cache-hit peeks are counted under RecentDataRequestRate). +LaggingConsumerRequestRateThrottledRate Rate of lagging consumer requests that had to wait on the request-rate limiter (cost/QPS protection) per second. Together with LaggingConsumerByteRateThrottledRate, attributes which limiter is the binding constraint (the combined wait is in LaggingConsumerRateLimitWaitTime). +LaggingConsumerRequestRejectedRate Rate of lagging consumer requests rejected due to executor unavailability per second +PartitionCorruptRecordRate Rate of partition responses returning CORRUPT_MESSAGE or INVALID_RECORD from FetchCompleter because a batch failed integrity validation (CRC / declared size, or a coalesced-run offset mismatch), with no valid prefix to serve. +PartitionPartialFetchRate Rate of partition responses that were truncated mid-batch-list due to a missing extent or validation failure on a trailing batch. Successful prefix is returned to the consumer; the trailing batches are dropped. +PartitionStorageErrorRate Rate of partition responses returning KAFKA_STORAGE_ERROR from FetchCompleter (missing batch metadata, or missing/incomplete extents so no records were constructable). Data-integrity failures are counted separately under PartitionCorruptRecordRate. +RecentDataRequestRate Rate of requests served via the hot path (recent data with cache) per second. Under the consolidation metrics group (ConsolidationFetchMetrics) this counts cache-hit peeks that reuse consumer-cached data. +======================================== ================================================================================================================================================================================================================================================================================================================================================================= InklessFetchOffset metrics diff --git a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java index 705054dd619..180ca2bd07d 100644 --- a/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java +++ b/server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java @@ -216,6 +216,14 @@ public class ServerConfigs { "for consolidation cold-path fetches. 0 (default) means unlimited -- consolidation is internal background work and does not " + "need throttling under normal conditions. Set > 0 as a safety valve to bound object storage request rate from consolidation."; + public static final String DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_CONFIG = "diskless.consolidation.fetch.lagging.byte.rate.limit"; + public static final long DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_DEFAULT = 0; + public static final String DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_DOC = "Maximum bytes per second read from object storage " + + "for consolidation cold-path fetches. This bounds the physical object-storage read bandwidth (the S3 GET range size, including read " + + "amplification from bounding-range reads over multi-partition objects), which is distinct from " + DISKLESS_CONSOLIDATION_FETCH_RATE_LIMIT_BYTES_PER_SECOND_CONFIG + " " + + "that limits the processed record bytes appended to the consolidated log. 0 (default) means unlimited. Set > 0 to protect the node's " + + "network bandwidth from consolidation read amplification."; + public static final String CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_CONFIG = "classic.remote.storage.force.enable"; public static final boolean CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_DEFAULT = false; public static final String CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_DOC = "Force classic topics to be created with remote.storage.enable=true, " + @@ -314,6 +322,8 @@ public class ServerConfigs { atLeast(0), LOW, DISKLESS_CONSOLIDATION_FETCH_RATE_LIMIT_BYTES_PER_SECOND_DOC) .define(DISKLESS_CONSOLIDATION_FETCH_LAGGING_REQUEST_RATE_LIMIT_CONFIG, INT, DISKLESS_CONSOLIDATION_FETCH_LAGGING_REQUEST_RATE_LIMIT_DEFAULT, atLeast(0), LOW, DISKLESS_CONSOLIDATION_FETCH_LAGGING_REQUEST_RATE_LIMIT_DOC) + .define(DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_CONFIG, LONG, DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_DEFAULT, + atLeast(0), LOW, DISKLESS_CONSOLIDATION_FETCH_LAGGING_BYTE_RATE_LIMIT_DOC) .define(CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_CONFIG, BOOLEAN, CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_DEFAULT, LOW, CLASSIC_REMOTE_STORAGE_FORCE_ENABLE_DOC) .define(CLASSIC_REMOTE_STORAGE_FORCE_EXCLUDE_TOPIC_REGEXES_CONFIG, LIST, CLASSIC_REMOTE_STORAGE_FORCE_EXCLUDE_TOPIC_REGEXES_DEFAULT, diff --git a/storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java b/storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java index eda350a0003..03f2f139175 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java @@ -23,6 +23,9 @@ import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.network.ListenerName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.lang.reflect.InvocationTargetException; import java.time.Duration; import java.util.Collections; @@ -37,6 +40,8 @@ import io.aiven.inkless.storage_backend.in_memory.InMemoryStorage; public class InklessConfig extends AbstractConfig { + + private static final Logger LOG = LoggerFactory.getLogger(InklessConfig.class); public static final String PREFIX = "inkless."; public static final String CONTROL_PLANE_PREFIX = "control.plane."; @@ -209,6 +214,19 @@ public class InklessConfig extends AbstractConfig { // Tune based on storage backend capacity and budget constraints. private static final int FETCH_LAGGING_CONSUMER_REQUEST_RATE_LIMIT_DEFAULT = 200; + public static final String FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_CONFIG = "fetch.lagging.consumer.byte.rate.limit"; + public static final String FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_DOC = "Maximum bytes per second read from object storage for lagging consumer data fetches. " + + "This caps the storage-to-broker (ingress) throughput of the cold path to keep it below the node's baseline network bandwidth. " + + "It is independent of and complementary to " + FETCH_LAGGING_CONSUMER_REQUEST_RATE_LIMIT_CONFIG + ": the request-rate limit " + + "protects against storage GET request cost (QPS), while this byte-rate limit protects network bandwidth. " + + "Set to 0 (default) to disable byte-rate limiting. " + + "Metered by the fetched byte range, so it governs storage-to-broker ingress only (not broker-to-consumer egress). " + + "A single cold fetch covers one storage object (bounding range), so it should be set at or above the produced object size " + + "(" + PRODUCE_BUFFER_MAX_BYTES_CONFIG + "); a lower value only throttles the stream of fetches, it cannot split an " + + "individual object below its own size. " + + "Note: hedge requests triggered by slow fetches are exempt from this limit."; + private static final long FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_DEFAULT = 0; + public static final String FETCH_HEDGE_TTFB_THRESHOLD_MS_CONFIG = "fetch.hedge.ttfb.threshold.ms"; public static final String FETCH_HEDGE_TTFB_THRESHOLD_MS_DOC = "Time-to-first-byte threshold in milliseconds to trigger a hedge request. " + "When a storage fetch has not received its first byte within this threshold, a competing hedge request is submitted. " @@ -468,6 +486,14 @@ public static ConfigDef configDef() { ConfigDef.Importance.MEDIUM, FETCH_LAGGING_CONSUMER_REQUEST_RATE_LIMIT_DOC ); + configDef.define( + FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_CONFIG, + ConfigDef.Type.LONG, + FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_DEFAULT, + ConfigDef.Range.atLeast(0), + ConfigDef.Importance.MEDIUM, + FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_DOC + ); configDef.define( FETCH_HEDGE_TOTAL_TIME_THRESHOLD_MS_CONFIG, ConfigDef.Type.LONG, @@ -672,6 +698,24 @@ private static ConfigDef validate(final Map props) { (List) parsedProps.get(CLIENT_AZ_LISTENER_MAP_CONFIG); parseClientAzListenerMap(azListenerEntries); + // Warn (do not reject) if the lagging byte-rate limit is below the produced object size. + // A single cold fetch covers one storage object (bounding range) and is indivisible, so a limit + // below the object size cannot throttle an individual fetch below its own size: such fetches are + // charged a full bucket and let through (tracked via LaggingConsumerByteRateOversizedRate). + // produce.buffer.max.bytes is only a best-effort, node-local estimate of object size, so this is + // a heuristic sanity check rather than a guarantee. + final long laggingByteRateLimit = + ((Number) parsedProps.get(FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_CONFIG)).longValue(); + final int produceBufferMaxBytes = + ((Number) parsedProps.get(PRODUCE_BUFFER_MAX_BYTES_CONFIG)).intValue(); + if (laggingByteRateLimit > 0 && laggingByteRateLimit < produceBufferMaxBytes) { + LOG.warn("{} ({} bytes/s) is below {} ({} bytes); individual cold fetches larger than the limit " + + "cannot be throttled and will pass through charged a full bucket. This is likely a " + + "misconfiguration - set the byte-rate limit at or above the produced object size.", + FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_CONFIG, laggingByteRateLimit, + PRODUCE_BUFFER_MAX_BYTES_CONFIG, produceBufferMaxBytes); + } + return configDef; } @@ -798,6 +842,10 @@ public int fetchLaggingConsumerRequestRateLimit() { return getInt(FETCH_LAGGING_CONSUMER_REQUEST_RATE_LIMIT_CONFIG); } + public long fetchLaggingConsumerByteRateLimit() { + return getLong(FETCH_LAGGING_CONSUMER_BYTE_RATE_LIMIT_CONFIG); + } + public long fetchHedgeTtfbThresholdMs() { return getLong(FETCH_HEDGE_TTFB_THRESHOLD_MS_CONFIG); } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java index ba8f7c27ce5..4317bfac19b 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchHandler.java @@ -58,6 +58,7 @@ public FetchHandler(final SharedState state) { state.maybeLaggingFetchStorage(), state.config().fetchLaggingConsumerThresholdMs(), state.config().fetchLaggingConsumerRequestRateLimit(), + state.config().fetchLaggingConsumerByteRateLimit(), state.config().fetchLaggingConsumerThreadPoolSize(), state.config().fetchHedgeTtfbThresholdMs(), state.config().fetchHedgeTotalTimeThresholdMs(), diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchPlanner.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchPlanner.java index 897a3fd1b45..4dff8aacbe9 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchPlanner.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchPlanner.java @@ -85,6 +85,8 @@ public class FetchPlanner implements Supplier submitSingleRequest(final ObjectFetchReque // This prevents hedges from firing while the primary is waiting for a rate limit token. final CompletableFuture fetchStarted = new CompletableFuture<>(); final CompletableFuture primary = CompletableFuture.supplyAsync(() -> { - // Apply rate limiting if configured (rate limit > 0) - if (laggingRateLimiter != null) { - applyRateLimit(); // InterruptedException here is wrapped in FetchException + // Apply rate limiting if configured (request-rate and/or byte-rate) + if (laggingRateLimiter != null || laggingByteRateLimiter != null) { + applyRateLimit(request.byteRange().size()); // InterruptedException here is wrapped in FetchException } // Signal that rate limiting is done and the fetch is starting. // Hedge timers begin counting from this point. @@ -490,13 +496,37 @@ private void tryFireHedge( } } - // Applies request-based rate limiting by blocking executor thread until token available. - // Always records wait time (including zero-wait) for accurate latency histogram. + // Applies rate limiting by blocking the executor thread until tokens are available. + // Two independent limiters may apply: request-rate (cost/QPS protection, 1 token per request) + // and byte-rate (bandwidth protection, byteRange size in tokens). Either may be disabled (null). + // Each limiter is acquired non-blocking first; only on failure do we record a throttle hit and + // block. This keeps the combined wait in a single histogram while still attributing which limiter + // was the binding constraint. Always records total wait time (including zero-wait) for an unbiased + // latency histogram. // Note: If interrupted, the duration is still recorded before the exception is thrown. - private void applyRateLimit() { + private void applyRateLimit(final long bytes) { TimeUtils.measureDurationMs(time, () -> { try { - laggingRateLimiter.asBlocking().consume(1); + if (laggingRateLimiter != null && !laggingRateLimiter.tryConsume(1)) { + metrics.recordRequestRateThrottled(); + laggingRateLimiter.asBlocking().consume(1); + } + if (laggingByteRateLimiter != null) { + // A single cold fetch covers one storage object and is indivisible. If it is + // larger than the bucket capacity it can never be fully satisfied, so charge a + // full bucket (the max the limiter can enforce) and let it through instead of + // blocking forever. This only happens when the byte-rate limit is set below the + // produced object size (likely misconfiguration), tracked via the oversized meter. + final long cost = Math.min(bytes, laggingByteRateCapacity); + if (bytes > laggingByteRateCapacity) { + metrics.recordByteRateOversized(); + } + // cost == 0 only for a degenerate empty range; skip since Bucket4j rejects non-positive consume. + if (cost > 0 && !laggingByteRateLimiter.tryConsume(cost)) { + metrics.recordByteRateThrottled(); + laggingByteRateLimiter.asBlocking().consume(cost); + } + } } catch (final InterruptedException e) { // Rate limit wait was interrupted (typically during shutdown). // Preserve interrupt status for executor framework, but wrap in FetchException diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java index bc6f004fad0..f735811206c 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java @@ -102,7 +102,17 @@ public class InklessFetchMetrics { // Always records wait time to avoid histogram bias - zero-wait cases show when rate limiting is NOT a bottleneck. // Use to monitor: rate limiting latency distribution, actual throttling pressure, and limiter effectiveness. private static final String LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME = "LaggingConsumerRateLimitWaitTime"; - private static final String LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME_DOC = "Wait time for rate-limited lagging consumer requests in milliseconds"; + private static final String LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME_DOC = "Combined wait time (request-rate plus byte-rate limiters) for rate-limited lagging consumer requests in milliseconds"; + + private static final String LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE = "LaggingConsumerRequestRateThrottledRate"; + private static final String LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE_DOC = "Rate of lagging consumer requests that had to wait on the request-rate limiter (cost/QPS protection) per second. " + + "Together with LaggingConsumerByteRateThrottledRate, attributes which limiter is the binding constraint (the combined wait is in LaggingConsumerRateLimitWaitTime)."; + private static final String LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE = "LaggingConsumerByteRateThrottledRate"; + private static final String LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE_DOC = "Rate of lagging consumer requests that had to wait on the byte-rate limiter (bandwidth protection) per second. " + + "Together with LaggingConsumerRequestRateThrottledRate, attributes which limiter is the binding constraint (the combined wait is in LaggingConsumerRateLimitWaitTime)."; + private static final String LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE = "LaggingConsumerByteRateOversizedRate"; + private static final String LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE_DOC = "Rate of lagging consumer fetches whose byte range exceeds the byte-rate limiter capacity (fetch.lagging.consumer.byte.rate.limit) per second. " + + "Such a fetch cannot be split below its own size, so it is charged a full bucket and allowed through. A non-zero rate indicates the byte-rate limit is set below the produced object size (likely misconfiguration)."; private static final String HEDGE_REQUEST_RATE = "HedgeRequestRate"; private static final String HEDGE_REQUEST_RATE_DOC = "Rate of hedged fetch requests issued per second"; @@ -147,6 +157,9 @@ public static List all() { new MetricNameTemplate(LAGGING_CONSUMER_REQUEST_RATE, GROUP, LAGGING_CONSUMER_REQUEST_RATE_DOC), new MetricNameTemplate(LAGGING_CONSUMER_REQUEST_REJECTED_RATE, GROUP, LAGGING_CONSUMER_REQUEST_REJECTED_RATE_DOC), new MetricNameTemplate(LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME, GROUP, LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME_DOC), + new MetricNameTemplate(LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE, GROUP, LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE_DOC), + new MetricNameTemplate(LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE, GROUP, LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE_DOC), + new MetricNameTemplate(LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE, GROUP, LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE_DOC), new MetricNameTemplate(HEDGE_REQUEST_RATE, GROUP, HEDGE_REQUEST_RATE_DOC), new MetricNameTemplate(HEDGE_TTFB_TRIGGERED_RATE, GROUP, HEDGE_TTFB_TRIGGERED_RATE_DOC), new MetricNameTemplate(HEDGE_TOTAL_TIME_TRIGGERED_RATE, GROUP, HEDGE_TOTAL_TIME_TRIGGERED_RATE_DOC), @@ -185,6 +198,9 @@ public static List all() { private final Meter laggingConsumerRequestRate; private final Meter laggingConsumerRejectedRate; private final Histogram laggingRateLimitWaitTime; + private final Meter laggingRequestRateThrottledRate; + private final Meter laggingByteRateThrottledRate; + private final Meter laggingByteRateOversizedRate; private final Meter hedgeRequestRate; private final Meter hedgeTtfbTriggeredRate; private final Meter hedgeTotalTimeTriggeredRate; @@ -225,6 +241,9 @@ public InklessFetchMetrics(final Time time, final ObjectCache cache, final Kafka laggingConsumerRequestRate = metricsGroup.newMeter(LAGGING_CONSUMER_REQUEST_RATE, "requests", TimeUnit.SECONDS, Map.of()); laggingConsumerRejectedRate = metricsGroup.newMeter(LAGGING_CONSUMER_REQUEST_REJECTED_RATE, "rejections", TimeUnit.SECONDS, Map.of()); laggingRateLimitWaitTime = metricsGroup.newHistogram(LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME, true, Map.of()); + laggingRequestRateThrottledRate = metricsGroup.newMeter(LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE, "requests", TimeUnit.SECONDS, Map.of()); + laggingByteRateThrottledRate = metricsGroup.newMeter(LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE, "requests", TimeUnit.SECONDS, Map.of()); + laggingByteRateOversizedRate = metricsGroup.newMeter(LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE, "requests", TimeUnit.SECONDS, Map.of()); hedgeRequestRate = metricsGroup.newMeter(HEDGE_REQUEST_RATE, "hedges", TimeUnit.SECONDS, Map.of()); hedgeTtfbTriggeredRate = metricsGroup.newMeter(HEDGE_TTFB_TRIGGERED_RATE, "hedges", TimeUnit.SECONDS, Map.of()); hedgeTotalTimeTriggeredRate = metricsGroup.newMeter(HEDGE_TOTAL_TIME_TRIGGERED_RATE, "hedges", TimeUnit.SECONDS, Map.of()); @@ -321,6 +340,9 @@ public void close() { metricsGroup.removeMetric(LAGGING_CONSUMER_REQUEST_RATE); metricsGroup.removeMetric(LAGGING_CONSUMER_REQUEST_REJECTED_RATE); metricsGroup.removeMetric(LAGGING_CONSUMER_RATE_LIMIT_WAIT_TIME); + metricsGroup.removeMetric(LAGGING_CONSUMER_REQUEST_RATE_THROTTLED_RATE); + metricsGroup.removeMetric(LAGGING_CONSUMER_BYTE_RATE_THROTTLED_RATE); + metricsGroup.removeMetric(LAGGING_CONSUMER_BYTE_RATE_OVERSIZED_RATE); metricsGroup.removeMetric(HEDGE_REQUEST_RATE); metricsGroup.removeMetric(HEDGE_TTFB_TRIGGERED_RATE); metricsGroup.removeMetric(HEDGE_TOTAL_TIME_TRIGGERED_RATE); @@ -440,6 +462,37 @@ public void recordRateLimitWaitTime(long waitMs) { laggingRateLimitWaitTime.update(waitMs); } + /** + * Records that a lagging consumer request had to wait on the request-rate limiter + * (cost/QPS protection). Call only when the non-blocking acquire failed and a blocking + * wait was required, so the meter reflects how often request-rate is the binding constraint. + * Metric: LaggingConsumerRequestRateThrottledRate + */ + public void recordRequestRateThrottled() { + laggingRequestRateThrottledRate.mark(); + } + + /** + * Records that a lagging consumer request had to wait on the byte-rate limiter + * (bandwidth protection). Call only when the non-blocking acquire failed and a blocking + * wait was required, so the meter reflects how often byte-rate is the binding constraint. + * Metric: LaggingConsumerByteRateThrottledRate + */ + public void recordByteRateThrottled() { + laggingByteRateThrottledRate.mark(); + } + + /** + * Records a lagging consumer fetch whose byte range exceeds the byte-rate limiter capacity. + * Such a fetch is charged a full bucket and allowed through (it cannot be split below its own + * size). A non-zero rate indicates fetch.lagging.consumer.byte.rate.limit is set below the + * produced object size (likely misconfiguration). + * Metric: LaggingConsumerByteRateOversizedRate + */ + public void recordByteRateOversized() { + laggingByteRateOversizedRate.mark(); + } + public void recordHedgeRequest() { hedgeRequestRate.mark(); } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java index d1de547d3ef..e457ecde2ce 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java @@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.time.Duration; import java.time.Instant; import java.util.List; import java.util.Map; @@ -119,6 +120,8 @@ public class Reader implements AutoCloseable { private final InklessFetchMetrics fetchMetrics; private final BrokerTopicStats brokerTopicStats; private final Bucket rateLimiter; + private final Bucket byteRateLimiter; + private final long byteRateCapacity; private ThreadPoolMonitor metadataThreadPoolMonitor; private ThreadPoolMonitor dataThreadPoolMonitor; private ThreadPoolMonitor laggingConsumerThreadPoolMonitor; @@ -136,6 +139,7 @@ public Reader( Optional maybeLaggingFetchStorage, long laggingConsumerThresholdMs, int laggingConsumerRequestRateLimit, + long laggingConsumerByteRateLimit, int laggingConsumerThreadPoolSize, long hedgeTtfbThresholdMs, long hedgeTotalTimeThresholdMs, @@ -143,7 +147,7 @@ public Reader( ) { this(time, objectKeyCreator, keyAlignmentStrategy, cache, controlPlane, objectFetcher, brokerTopicStats, fetchMetadataThreadPoolSize, fetchDataThreadPoolSize, maybeLaggingFetchStorage, - laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, laggingConsumerThreadPoolSize, + laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, laggingConsumerByteRateLimit, laggingConsumerThreadPoolSize, hedgeTtfbThresholdMs, hedgeTotalTimeThresholdMs, maxBatchesPerPartitionToFind, new KafkaMetricsGroup(InklessFetchMetrics.class.getPackageName(), InklessFetchMetrics.class.getSimpleName()), @@ -163,6 +167,7 @@ public Reader( Optional maybeLaggingFetchStorage, long laggingConsumerThresholdMs, int laggingConsumerRequestRateLimit, + long laggingConsumerByteRateLimit, int laggingConsumerThreadPoolSize, long hedgeTtfbThresholdMs, long hedgeTotalTimeThresholdMs, @@ -183,6 +188,7 @@ public Reader( maybeLaggingFetchStorage, laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, + laggingConsumerByteRateLimit, laggingConsumerThreadPoolSize, hedgeTtfbThresholdMs, hedgeTotalTimeThresholdMs, @@ -206,6 +212,7 @@ public Reader( Optional maybeLaggingFetchStorage, long laggingConsumerThresholdMs, int laggingConsumerRequestRateLimit, + long laggingConsumerByteRateLimit, int laggingConsumerThreadPoolSize, long hedgeTtfbThresholdMs, long hedgeTotalTimeThresholdMs, @@ -239,6 +246,7 @@ public Reader( : maybeLaggingFetchStorage.orElse(null), laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, + laggingConsumerByteRateLimit, // Only create a dedicated lagging pool when pool size > 0. // When pool size is 0 and a fetcher is present, the cold path reuses fetchDataExecutor. laggingConsumerThreadPoolSize > 0 @@ -313,7 +321,7 @@ private static ExecutorService createBoundedThreadPool(String threadNamePrefix, ) { this(time, objectKeyCreator, keyAlignmentStrategy, cache, controlPlane, objectFetcher, maxBatchesPerPartitionToFind, metadataExecutor, fetchDataExecutor, - laggingConsumerObjectFetcher, laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, + laggingConsumerObjectFetcher, laggingConsumerThresholdMs, laggingConsumerRequestRateLimit, 0L, laggingFetchDataExecutor, hedgeScheduler, hedgeTtfbThresholdMs, hedgeTotalTimeThresholdMs, fetchMetrics, brokerTopicStats, monitorPrefix, false); } @@ -332,6 +340,7 @@ private static ExecutorService createBoundedThreadPool(String threadNamePrefix, ObjectFetcher laggingConsumerObjectFetcher, long laggingConsumerThresholdMs, int laggingConsumerRequestRateLimit, + long laggingConsumerByteRateLimit, ExecutorService laggingFetchDataExecutor, ScheduledExecutorService hedgeScheduler, long hedgeTtfbThresholdMs, @@ -398,7 +407,7 @@ private static ExecutorService createBoundedThreadPool(String threadNamePrefix, // pool limits concurrent execution, providing both rate limiting and concurrency control. final Bandwidth limit = Bandwidth.builder() .capacity(laggingConsumerRequestRateLimit) - .refillGreedy(laggingConsumerRequestRateLimit, java.time.Duration.ofSeconds(1)) + .refillGreedy(laggingConsumerRequestRateLimit, Duration.ofSeconds(1)) .build(); this.rateLimiter = Bucket.builder() .addLimit(limit) @@ -407,6 +416,26 @@ private static ExecutorService createBoundedThreadPool(String threadNamePrefix, this.rateLimiter = null; } + // Byte-rate limiter: bounds cold-path storage-to-broker bandwidth per node. Independent of the + // request-rate limiter above (that protects storage GET cost; this protects network bandwidth). + // capacity = byteRateLimit: a 1-second burst, consistent with the request limiter. A single cold + // fetch covers one storage object and is indivisible; if it exceeds capacity it can never be fully + // satisfied, so FetchPlanner charges a full bucket and lets it through (see applyRateLimit) rather + // than deadlocking. That only occurs when the limit is set below the produced object size. + if (this.laggingFetchDataExecutor != null && laggingConsumerByteRateLimit > 0) { + final Bandwidth byteLimit = Bandwidth.builder() + .capacity(laggingConsumerByteRateLimit) + .refillGreedy(laggingConsumerByteRateLimit, Duration.ofSeconds(1)) + .build(); + this.byteRateLimiter = Bucket.builder() + .addLimit(byteLimit) + .build(); + this.byteRateCapacity = laggingConsumerByteRateLimit; + } else { + this.byteRateLimiter = null; + this.byteRateCapacity = 0L; + } + this.isConsolidationFetch = isConsolidationFetch; this.fetchMetrics = fetchMetrics; this.brokerTopicStats = brokerTopicStats; @@ -490,6 +519,8 @@ public CompletableFuture> fetch( laggingConsumerObjectFetcher, laggingConsumerThresholdMs, rateLimiter, + byteRateLimiter, + byteRateCapacity, laggingFetchDataExecutor, hedgeScheduler, hedgeTtfbThresholdMs, diff --git a/storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java b/storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java index 23410dde303..e29c4b85227 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java @@ -317,6 +317,9 @@ void laggingConsumerConfigDefaults() { // Default rate limit assertThat(config.fetchLaggingConsumerRequestRateLimit()).isEqualTo(200); + // Default byte-rate limit: 0 (disabled) + assertThat(config.fetchLaggingConsumerByteRateLimit()).isEqualTo(0L); + // Default threshold: -1 (auto) should use heuristic: cache TTL // Default cache TTL is 60 seconds, so threshold should follow that assertThat(config.fetchLaggingConsumerThresholdMs()).isEqualTo(60_000L); @@ -330,12 +333,14 @@ void laggingConsumerConfigExplicitValues() { configs.put("storage.backend.class", ConfigTestStorageBackend.class.getCanonicalName()); configs.put("fetch.lagging.consumer.thread.pool.size", "32"); configs.put("fetch.lagging.consumer.request.rate.limit", "500"); + configs.put("fetch.lagging.consumer.byte.rate.limit", "104857600"); // 100 MiB/s configs.put("fetch.lagging.consumer.threshold.ms", "300000"); // 5 minutes explicit final var config = new InklessConfig(configs); assertThat(config.fetchLaggingConsumerThreadPoolSize()).isEqualTo(32); assertThat(config.fetchLaggingConsumerRequestRateLimit()).isEqualTo(500); + assertThat(config.fetchLaggingConsumerByteRateLimit()).isEqualTo(104857600L); assertThat(config.fetchLaggingConsumerThresholdMs()).isEqualTo(300_000L); } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java index 433b54065ba..13bd4707b96 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java @@ -34,6 +34,7 @@ import java.nio.ByteBuffer; import java.nio.channels.ReadableByteChannel; +import java.time.Duration; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -71,6 +72,7 @@ import io.aiven.inkless.generated.CacheKey; import io.aiven.inkless.generated.FileExtent; import io.aiven.inkless.storage_backend.common.ObjectFetcher; +import io.aiven.inkless.test_utils.SynchronousExecutor; import io.github.bucket4j.Bucket; import static org.assertj.core.api.Assertions.assertThat; @@ -786,7 +788,7 @@ public void recentDataUsesRecentExecutorWithoutRateLimit() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { final byte[] expectedData = "recent-data".getBytes(); final Bucket rateLimiter = Bucket.builder() - .addLimit(limit -> limit.capacity(1).refillGreedy(1, java.time.Duration.ofSeconds(1))) + .addLimit(limit -> limit.capacity(1).refillGreedy(1, Duration.ofSeconds(1))) .build(); when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); @@ -868,7 +870,7 @@ public void laggingDataUsesLaggingExecutorWithRateLimit() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { final byte[] expectedData = "old-data".getBytes(); final Bucket rateLimiter = Bucket.builder() - .addLimit(limit -> limit.capacity(10).refillGreedy(10, java.time.Duration.ofSeconds(1))) + .addLimit(limit -> limit.capacity(10).refillGreedy(10, Duration.ofSeconds(1))) .build(); when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); @@ -1051,6 +1053,8 @@ public void executorQueueFullThrowsRejectedExecutionException() throws Exception fetcher, // laggingConsumerFetcher threshold, null, // No rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity saturatedExecutor, // Saturated executor null, // no hedge scheduler 0, // TTFB hedging disabled @@ -1131,6 +1135,8 @@ public void executorShutdownTrackedAsRejection() throws Exception { fetcher, // laggingConsumerFetcher threshold, null, // No rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity shutdownExecutor, // Shutdown executor null, // no hedge scheduler 0, // TTFB hedging disabled @@ -1255,6 +1261,8 @@ public void hotAndColdPathsExecuteConcurrently() throws Exception { fetcher, // laggingConsumerFetcher threshold, null, // No rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity coldExecutor, // Cold path executor null, // no hedge scheduler 0, // TTFB hedging disabled @@ -1377,7 +1385,7 @@ public void bothFeaturesCanBeEnabled() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { final byte[] expectedData = "cold-with-limit".getBytes(); final Bucket rateLimiter = Bucket.builder() - .addLimit(limit -> limit.capacity(10).refillGreedy(10, java.time.Duration.ofSeconds(1))) + .addLimit(limit -> limit.capacity(10).refillGreedy(10, Duration.ofSeconds(1))) .build(); when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); @@ -1608,6 +1616,8 @@ private FetchPlanner createFetchPlannerWithCustomThreshold( fetcher, thresholdMs, rateLimiter, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingConsumerExecutor, null, // no hedge scheduler 0, // TTFB hedging disabled @@ -1619,6 +1629,44 @@ private FetchPlanner createFetchPlannerWithCustomThreshold( ); } + /** + * Creates a FetchPlanner with a byte-rate limiter (and optionally a request-rate limiter). + * Use for testing byte-rate (bandwidth) limiting on the cold path. + */ + private FetchPlanner createFetchPlannerWithByteRateLimit( + KeyAlignmentStrategy keyAlignmentStrategy, + ObjectCache cache, + Map batchCoordinatesFuture, + long thresholdMs, + ExecutorService laggingConsumerExecutor, + Bucket rateLimiter, + Bucket byteRateLimiter, + long byteRateCapacity + ) { + return new FetchPlanner( + time, + FetchPlannerTest.OBJECT_KEY_CREATOR, + keyAlignmentStrategy, + cache, + fetcher, + fetchDataExecutor, + fetcher, + thresholdMs, + rateLimiter, + byteRateLimiter, + byteRateCapacity, + laggingConsumerExecutor, + null, // no hedge scheduler + 0, // TTFB hedging disabled + 0, // total-time hedging disabled + new ConcurrentHashMap<>(), + batchCoordinatesFuture, + false, + metrics + ); + } + + /** * Creates a FetchPlanner with lagging consumer feature enabled (default 60s threshold). * Use for testing hot/cold path separation with standard threshold. @@ -1657,6 +1705,165 @@ private void assertBatchPlan(Map coordinate assertThat(new HashSet<>(actualJobs)).isEqualTo(expectedJobs); } + @Nested + class ByteRateLimitTests { + + private Map coldCoordinates(final long byteSize) { + final long oldTimestamp = time.milliseconds() - 120_000L; // 2 minutes ago -> cold path + return Map.of( + partition0, FindBatchResponse.success(List.of( + new BatchInfo(1L, OBJECT_KEY_A.value(), + BatchMetadata.of(partition0, 0, byteSize, 0, 0, 10, oldTimestamp, TimestampType.CREATE_TIME)) + ), 0, 1) + ); + } + + @Test + public void byteRateLimitedFetchCompletesAndRecordsWaitTime() throws Exception { + try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { + final byte[] expectedData = "cold-data".getBytes(); + // Generous capacity: the fetch is not throttled, but the byte limiter path is exercised. + final long capacity = 1_000_000L; + final Bucket byteRateLimiter = Bucket.builder() + .addLimit(limit -> limit.capacity(capacity).refillGreedy(capacity, Duration.ofSeconds(1))) + .build(); + + when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); + when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + + final FetchPlanner planner = createFetchPlannerWithByteRateLimit( + keyAlignmentStrategy, caffeineCache, coldCoordinates(100), + 60 * 1000L, laggingFetchDataExecutor, null, byteRateLimiter, capacity + ); + + final FileExtent result = planner.get().get(0).future().get(); + assertThat(result.data()).isEqualTo(expectedData); + + verify(metrics).recordLaggingConsumerRequest(); + verify(metrics).recordRateLimitWaitTime(any(Long.class)); + verify(metrics, never()).recordByteRateOversized(); + } + } + + @Test + public void oversizedFetchIsChargedFullBucketAndCompletes() throws Exception { + try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { + final byte[] expectedData = "oversized-cold-data".getBytes(); + // Capacity smaller than the fetched byte range: the fetch can never be fully satisfied, + // so it is charged a full bucket and let through rather than deadlocking. + final long capacity = 10L; + final long fetchByteSize = 1_000L; + final Bucket byteRateLimiter = Bucket.builder() + .addLimit(limit -> limit.capacity(capacity).refillGreedy(capacity, Duration.ofSeconds(1))) + .build(); + + when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); + when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + + final FetchPlanner planner = createFetchPlannerWithByteRateLimit( + keyAlignmentStrategy, caffeineCache, coldCoordinates(fetchByteSize), + 60 * 1000L, laggingFetchDataExecutor, null, byteRateLimiter, capacity + ); + + // Must complete (not hang) even though the fetch exceeds the bucket capacity. + final FileExtent result = planner.get().get(0).future().get(5, TimeUnit.SECONDS); + assertThat(result.data()).isEqualTo(expectedData); + + verify(metrics).recordByteRateOversized(); + verify(metrics).recordRateLimitWaitTime(any(Long.class)); + } + } + + @Test + public void byteRateLimitEnabledIndependentlyOfRequestRate() throws Exception { + try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { + final byte[] expectedData = "byte-only".getBytes(); + final long capacity = 1_000_000L; + final Bucket byteRateLimiter = Bucket.builder() + .addLimit(limit -> limit.capacity(capacity).refillGreedy(capacity, Duration.ofSeconds(1))) + .build(); + + when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); + when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + + // request-rate limiter is null; only the byte-rate limiter is configured. + final FetchPlanner planner = createFetchPlannerWithByteRateLimit( + keyAlignmentStrategy, caffeineCache, coldCoordinates(100), + 60 * 1000L, laggingFetchDataExecutor, null, byteRateLimiter, capacity + ); + + final FileExtent result = planner.get().get(0).future().get(); + assertThat(result.data()).isEqualTo(expectedData); + + verify(metrics).recordLaggingConsumerRequest(); + verify(metrics).recordRateLimitWaitTime(any(Long.class)); + } + } + + @Test + public void byteRateThrottleBlocksWhenBucketDrained() throws Exception { + try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { + final byte[] expectedData = "byte-throttled".getBytes(); + final long capacity = 1_000_000L; + final long fetchByteSize = 200_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(); + + when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); + when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + + // Run the cold-path task inline (no executor scheduling gap), so applyRateLimit runs + // immediately after the drain: tryConsume(cost) deterministically fails and blocks on + // refill regardless of machine load. This removes the timing dependence. + final FetchPlanner planner = createFetchPlannerWithByteRateLimit( + keyAlignmentStrategy, caffeineCache, coldCoordinates(fetchByteSize), + 60 * 1000L, new SynchronousExecutor(), null, byteRateLimiter, capacity + ); + + final FileExtent result = planner.get().get(0).future().get(10, TimeUnit.SECONDS); + assertThat(result.data()).isEqualTo(expectedData); + + // The byte limiter blocked: throttle recorded, not oversized (cost <= capacity). + verify(metrics).recordByteRateThrottled(); + verify(metrics, never()).recordByteRateOversized(); + verify(metrics).recordRateLimitWaitTime(any(Long.class)); + } + } + + @Test + public void requestRateThrottleBlocksWhenBucketDrained() throws Exception { + try (CaffeineCache caffeineCache = new CaffeineCache(100, 0, 3600, 180)) { + final byte[] expectedData = "request-throttled".getBytes(); + final int capacity = 10; + 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(); + + when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(mock(ReadableByteChannel.class)); + when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + + // Run the cold-path task inline (no executor scheduling gap): tryConsume(1) runs + // immediately after the drain and deterministically fails, so it blocks on refill. + // Only the request-rate limiter is configured (byte-rate disabled). + final FetchPlanner planner = createFetchPlannerWithCustomThreshold( + keyAlignmentStrategy, caffeineCache, coldCoordinates(100), + 60 * 1000L, new SynchronousExecutor(), rateLimiter + ); + + final FileExtent result = planner.get().get(0).future().get(10, TimeUnit.SECONDS); + assertThat(result.data()).isEqualTo(expectedData); + + verify(metrics).recordRequestRateThrottled(); + verify(metrics).recordRateLimitWaitTime(any(Long.class)); + } + } + } + @Nested class HedgingTests { private final long hedgeThresholdMs = 50; @@ -1683,6 +1890,8 @@ private FetchPlanner createHedgingPlanner( fetcher, 60 * 1000L, null, // no rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, 0, // TTFB hedging disabled by default @@ -1766,6 +1975,8 @@ void hedgeTriggeredWhenPrimaryIsSlow() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, 0, // TTFB hedging disabled @@ -1870,6 +2081,8 @@ void hedgeRejectedWhenExecutorFull() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, 0, // TTFB hedging disabled @@ -1935,6 +2148,8 @@ void primaryFailsFastErrorPropagatesBeforeHedge() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, 0, // TTFB hedging disabled @@ -1999,6 +2214,8 @@ void bothFailPrimaryErrorPropagates() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, 0, // TTFB hedging disabled @@ -2069,6 +2286,8 @@ void hedgeTriggeredOnColdPath() throws Exception { fetcher, 60 * 1000L, // lagging threshold null, // no rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity multiExecutor, // cold path executor hedgeScheduler, 0, // TTFB hedging disabled @@ -2145,6 +2364,8 @@ void hedgeTriggeredByTtfbThreshold() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, ttfbThreshold, @@ -2210,6 +2431,8 @@ void ttfbHedgeNotTriggeredWhenFirstByteArrived() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, ttfbThreshold, @@ -2272,6 +2495,8 @@ void totalTimeHedgeFiresWhenTtfbDisabled() throws Exception { fetcher, 60 * 1000L, null, + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, hedgeScheduler, ttfbThreshold, @@ -2348,12 +2573,12 @@ void concurrentCallersOnSameKeyDedupToOneHedge() throws Exception { // Two planners sharing the same cache and metrics — simulates two concurrent Reader.fetch() calls final FetchPlanner planner1 = new FetchPlanner( time, OBJECT_KEY_CREATOR, keyAlignmentStrategy, cache, fetcher, - multiExecutor, fetcher, 60 * 1000L, null, laggingFetchDataExecutor, + multiExecutor, fetcher, 60 * 1000L, null, null, 0L, laggingFetchDataExecutor, hedgeScheduler, 0, hedgeThresholdMs, hedgeGuards, coordinates, false, metrics ); final FetchPlanner planner2 = new FetchPlanner( time, OBJECT_KEY_CREATOR, keyAlignmentStrategy, cache, fetcher, - multiExecutor, fetcher, 60 * 1000L, null, laggingFetchDataExecutor, + multiExecutor, fetcher, 60 * 1000L, null, null, 0L, laggingFetchDataExecutor, hedgeScheduler, 0, hedgeThresholdMs, hedgeGuards, coordinates, false, metrics ); @@ -2433,6 +2658,8 @@ void hedgeNotTriggeredDuringRateLimitWait() throws Exception { fetcher, 60 * 1000L, // lagging threshold rateLimiter, // rate limiter that blocks ~200ms + null, // no byte-rate limiter + 0L, // byte-rate capacity multiExecutor, // cold path executor hedgeScheduler, 0, // TTFB hedging disabled @@ -2492,7 +2719,7 @@ void hedgeStillFiresWhenFetchIsSlowAfterRateLimit() throws Exception { // Generous rate limiter (won't block significantly) final Bucket rateLimiter = Bucket.builder() - .addLimit(limit -> limit.capacity(10).refillGreedy(10, java.time.Duration.ofSeconds(1))) + .addLimit(limit -> limit.capacity(10).refillGreedy(10, Duration.ofSeconds(1))) .build(); final ExecutorService multiExecutor = Executors.newFixedThreadPool(2); @@ -2507,6 +2734,8 @@ void hedgeStillFiresWhenFetchIsSlowAfterRateLimit() throws Exception { fetcher, 60 * 1000L, rateLimiter, // non-blocking rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity multiExecutor, hedgeScheduler, 0, // TTFB disabled @@ -2743,6 +2972,8 @@ private FetchPlanner createConsolidationFetchPlanner( fetcher, 60 * 1000L, null, // no rate limiter + null, // no byte-rate limiter + 0L, // byte-rate capacity laggingFetchDataExecutor, null, // no hedge scheduler 0, // TTFB hedging disabled diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java index 1e4b6cf8d2c..6621153a435 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java @@ -1262,7 +1262,7 @@ void consolidationFetchWithoutFetcherThrows() { consolidationDataExecutor, // metadataExecutor consolidationDataExecutor, // fetchDataExecutor null, // laggingConsumerObjectFetcher — MISSING - 60_000L, 0, + 60_000L, 0, 0L, // byte-rate limit disabled null, // laggingFetchDataExecutor absent, so guard 1 is skipped null, 0, 0, coldPathMetrics, new BrokerTopicStats(), "consolidation-", @@ -1315,6 +1315,7 @@ void poolSizeZeroWithEmptyStorageIsAccepted() throws Exception { Optional.empty(), 60_000L, // laggingConsumerThresholdMs 0, // laggingConsumerRequestRateLimit + 0L, // byte-rate limit disabled 0, // laggingConsumerThreadPoolSize — feature disabled 0, // hedgeTtfbThresholdMs — disabled 0, // hedgeTotalTimeThresholdMs — disabled @@ -1339,6 +1340,7 @@ void poolSizePositiveWithStorageIsAccepted() throws Exception { Optional.of(mock(StorageBackend.class)), 60_000L, // laggingConsumerThresholdMs 0, // laggingConsumerRequestRateLimit + 0L, // byte-rate limit disabled 4, // laggingConsumerThreadPoolSize — feature enabled 0, // hedgeTtfbThresholdMs — disabled 0, // hedgeTotalTimeThresholdMs — disabled @@ -1365,6 +1367,7 @@ void poolSizeZeroWithStorageIsAccepted() throws Exception { Optional.of(mock(StorageBackend.class)), 60_000L, // laggingConsumerThresholdMs 0, // laggingConsumerRequestRateLimit + 0L, // byte-rate limit disabled 0, // laggingConsumerThreadPoolSize — no dedicated pool, reuses data pool 0, // hedgeTtfbThresholdMs — disabled 0, // hedgeTotalTimeThresholdMs — disabled @@ -1388,6 +1391,7 @@ void poolSizePositiveWithEmptyStorageThrows() { Optional.empty(), // no storage provided 60_000L, // laggingConsumerThresholdMs 0, // laggingConsumerRequestRateLimit + 0L, // byte-rate limit disabled 4, // laggingConsumerThreadPoolSize — feature enabled but no storage 0, // hedgeTtfbThresholdMs — disabled 0, // hedgeTotalTimeThresholdMs — disabled