fix(inkless:retention): size retention enforcement to keep up with produce

docs/inkless/configs.rst

@@ -258,18 +258,18 @@ Under ``inkless.``
   * Importance: low
 
 ``retention.enforcement.interval.ms``
-  The interval with which to enforce retention policies on a partition. This interval is approximate, because each scheduling event is randomized. The retention enforcement mechanism also takes into account the total number of brokers in the cluster: the more brokers, the less frequently each one of them enforces retention policy.
+  The interval with which to enforce retention policies on a partition. This interval is approximate, because each scheduling event is randomized. The retention enforcement mechanism also takes into account the total number of brokers in the cluster: the more brokers, the less frequently each one of them enforces retention policy. Together with retention.enforcement.max.batches.per.request this sets the sustained deletion capacity per partition (that many batches per this interval), which must exceed the rate at which batches are created for a partition or retention falls behind.
 
   * Type: int
-  * Default: 300000 (5 minutes)
+  * Default: 60000 (1 minute)
   * Valid Values: [1,...]
   * Importance: low
 
 ``retention.enforcement.max.batches.per.request``
-  The maximum number of batches to delete per partition when enforcing retention. A value of 0 means all eligible batches are deleted in one request, which makes the retention boundary scan proportional to the partition depth and can exceed the control-plane socket timeout on very deep partitions. A positive value bounds both the boundary scan and the delete to that many batches per pass, so a deep backlog drains over successive enforcement cycles instead of one unbounded request. It should be set above the per-interval expiry rate so retention does not fall behind.
+  The maximum number of batches to delete per partition when enforcing retention. A value of 0 means all eligible batches are deleted in one request, which makes the retention boundary scan proportional to the partition depth and can exceed the control-plane socket timeout on very deep partitions. A positive value bounds both the boundary scan and the delete to that many batches per pass, so a deep backlog drains over successive enforcement cycles instead of one unbounded request. It should be set above the per-interval expiry rate so retention does not fall behind: sustained deletion capacity per partition is this value per retention.enforcement.interval.ms, and batches are created for a partition at roughly one per produce.commit.interval.ms per broker writing to it.
 
   * Type: int
-  * Default: 1000
+  * Default: 2000
   * Valid Values: [0,...]
   * Importance: low
 

storage/inkless/src/main/java/io/aiven/inkless/config/InklessConfig.java

@@ -124,12 +124,17 @@ public class InklessConfig extends AbstractConfig {
     private static final String RETENTION_ENFORCEMENT_INTERVAL_MS_DOC = "The interval with which to enforce retention policies on a partition. " +
         "This interval is approximate, because each scheduling event is randomized. " +
         "The retention enforcement mechanism also takes into account the total number of brokers in the cluster: " +
-        "the more brokers, the less frequently each one of them enforces retention policy.";
-    private static final int RETENTION_ENFORCEMENT_INTERVAL_MS_DEFAULT = 5 * 60 * 1000;  // 5 minutes
+        "the more brokers, the less frequently each one of them enforces retention policy. " +
+        "Together with retention.enforcement.max.batches.per.request this sets the sustained deletion capacity " +
+        "per partition (that many batches per this interval), which must exceed the rate at which batches are " +
+        "created for a partition or retention falls behind.";
+    private static final int RETENTION_ENFORCEMENT_INTERVAL_MS_DEFAULT = 60 * 1000;  // 1 minute
 
     public static final String CONSOLIDATION_CLEANUP_INTERVAL_MS_CONFIG = "consolidation.cleanup.interval.ms";
     private static final String CONSOLIDATION_CLEANUP_INTERVAL_MS_DOC = "The interval with which to run consolidated diskless WAL pruning on each broker.";
-    private static final int CONSOLIDATION_CLEANUP_INTERVAL_MS_DEFAULT = RETENTION_ENFORCEMENT_INTERVAL_MS_DEFAULT;
+    // Deliberately not derived from RETENTION_ENFORCEMENT_INTERVAL_MS_DEFAULT: WAL pruning has its own cost
+    // profile (whole-file scans, not per-partition boundary scans), so it must not follow retention's cadence.
+    private static final int CONSOLIDATION_CLEANUP_INTERVAL_MS_DEFAULT = 5 * 60 * 1000;  // 5 minutes
 
     public static final String FILE_CLEANER_INTERVAL_MS_CONFIG = "file.cleaner.interval.ms";
     private static final String FILE_CLEANER_INTERVAL_MS_DOC = "The interval with which to clean up files marked for deletion.";
@@ -244,8 +249,10 @@ public class InklessConfig extends AbstractConfig {
         + "scan proportional to the partition depth and can exceed the control-plane socket timeout on very deep "
         + "partitions. A positive value bounds both the boundary scan and the delete to that many batches per pass, "
         + "so a deep backlog drains over successive enforcement cycles instead of one unbounded request. "
-        + "It should be set above the per-interval expiry rate so retention does not fall behind.";
-    private static final int RETENTION_ENFORCEMENT_MAX_BATCHES_PER_REQUEST_DEFAULT = 1000;
+        + "It should be set above the per-interval expiry rate so retention does not fall behind: sustained "
+        + "deletion capacity per partition is this value per retention.enforcement.interval.ms, and batches are "
+        + "created for a partition at roughly one per produce.commit.interval.ms per broker writing to it.";
+    private static final int RETENTION_ENFORCEMENT_MAX_BATCHES_PER_REQUEST_DEFAULT = 2000;
 
     public static final String BATCH_COALESCING_ENABLED_CONFIG = CONTROL_PLANE_PREFIX + "batch.coalescing.enabled";
     public static final String BATCH_COALESCING_ENABLED_DOC = "When true, contiguous same-partition batch runs within a single commit are collapsed "

storage/inkless/src/test/java/io/aiven/inkless/config/InklessConfigTest.java

@@ -109,7 +109,7 @@ void minimalConfig() {
         assertThat(config.produceUploadThreadPoolSize()).isEqualTo(8);
         assertThat(config.fetchDataThreadPoolSize()).isEqualTo(32);
         assertThat(config.fetchMetadataThreadPoolSize()).isEqualTo(8);
-        assertThat(config.maxBatchesPerEnforcementRequest()).isEqualTo(1000);
+        assertThat(config.maxBatchesPerEnforcementRequest()).isEqualTo(2000);
         assertThat(config.consolidationCleanupInterval()).isEqualTo(Duration.ofMinutes(5));
     }