aiven · jeqo · Jul 22, 2026
diff --git 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
@@ -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
@@ -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
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
@@ -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
@@ -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<String, ?> props) {
             (List<String>) 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
@@ -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
@@ -85,6 +85,8 @@ public class FetchPlanner implements Supplier<List<FetchPlanner.FetchRequestWith
     private final ObjectFetcher laggingObjectFetcher;
     private final long laggingConsumerThresholdMs;
     private final Bucket laggingRateLimiter;
+    private final Bucket laggingByteRateLimiter;
+    private final long laggingByteRateCapacity;
     private final ScheduledExecutorService hedgeScheduler;
     private final long hedgeTtfbThresholdMs;
     private final long hedgeTotalTimeThresholdMs;
@@ -103,6 +105,8 @@ public FetchPlanner(
         ObjectFetcher laggingObjectFetcher,
         long laggingConsumerThresholdMs,
         Bucket laggingRateLimiter,
+        Bucket laggingByteRateLimiter,
+        long laggingByteRateCapacity,
         ExecutorService laggingFetchDataExecutor,
         ScheduledExecutorService hedgeScheduler,
         long hedgeTtfbThresholdMs,
@@ -122,6 +126,8 @@ public FetchPlanner(
         this.laggingFetchDataExecutor = laggingFetchDataExecutor;
         this.laggingConsumerThresholdMs = laggingConsumerThresholdMs;
         this.laggingRateLimiter = laggingRateLimiter;
+        this.laggingByteRateLimiter = laggingByteRateLimiter;
+        this.laggingByteRateCapacity = laggingByteRateCapacity;
         this.hedgeScheduler = hedgeScheduler;
         this.hedgeTtfbThresholdMs = hedgeTtfbThresholdMs;
         this.hedgeTotalTimeThresholdMs = hedgeTotalTimeThresholdMs;
@@ -308,9 +314,9 @@ private CompletableFuture<FileExtent> submitSingleRequest(final ObjectFetchReque
                 // This prevents hedges from firing while the primary is waiting for a rate limit token.
                 final CompletableFuture<Void> fetchStarted = new CompletableFuture<>();
                 final CompletableFuture<FileExtent> 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