giuseppelillo · Sep 4, 2026 · Sep 3, 2026 · Sep 4, 2026 · Sep 4, 2026
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -21,7 +21,7 @@ import io.aiven.inkless.common.SharedState
 import io.aiven.inkless.consume.{ConcatenatedRecords, FetchHandler, FetchOffsetHandler, Reader}
 import io.aiven.inkless.storage_backend.common.ObjectFetcher
 import io.aiven.inkless.control_plane.{AdvanceCrossTierLogStartOffsetRequest, AdvanceCrossTierLogStartOffsetResponse, BatchInfo, FindBatchRequest, FindBatchResponse, InitDisklessLogProducerState, RepairDisklessLogRequest, ListOffsetsRequest => CpListOffsetsRequest}
-import io.aiven.inkless.delete.{DeleteRecordsInterceptor, FileCleaner, RetentionEnforcer}
+import io.aiven.inkless.delete.{DeleteRecordsInterceptor, FileCleaner, RetentionEnforcer, TopicPurger}
 import io.aiven.inkless.produce.AppendHandler
 import io.aiven.inkless.consolidation.{ConsolidatedDisklessLogPruner, ConsolidationFetcherManager, ConsolidationMetrics, ConsolidationReconciler, DelayedConsolidationFetch}
 import kafka.cluster.Partition
@@ -275,6 +275,7 @@ class ReplicaManager(val config: KafkaConfig,
   private val inklessDeleteRecordsInterceptor: Option[DeleteRecordsInterceptor] = inklessSharedState.map(new DeleteRecordsInterceptor(_))
   private val inklessRetentionEnforcer: Option[RetentionEnforcer] = inklessSharedState.map(new RetentionEnforcer(_))
   private val inklessFileCleaner: Option[FileCleaner] = inklessSharedState.map(new FileCleaner(_))
+  private val inklessTopicPurger: Option[TopicPurger] = inklessSharedState.map(new TopicPurger(_))
 
   // --- Diskless Partition Consolidation Fields ---
   private val inklessConsolidatedDisklessLogPruner: Option[ConsolidatedDisklessLogPruner] =
@@ -522,6 +523,8 @@ class ReplicaManager(val config: KafkaConfig,
 
       scheduler.schedule("inkless-file-cleaner", () => inklessFileCleaner.foreach(_.run()), sharedState.config().fileCleanerInterval().toMillis, sharedState.config().fileCleanerInterval().toMillis)
 
+      scheduler.schedule("inkless-topic-purger", () => inklessTopicPurger.foreach(_.run()), sharedState.config().topicPurgerInterval().toMillis, sharedState.config().topicPurgerInterval().toMillis)
+
       // The default 30s task delay would leave EARLIEST wrong for up to 30s after every startup.
       scheduler.schedule("inkless-cross-tier-log-start-reporter", () => sharedState.crossTierLogStartReporter().run(), sharedState.config().crossTierLogStartReportInterval().toMillis, sharedState.config().crossTierLogStartReportInterval().toMillis)
 
@@ -3548,6 +3551,7 @@ class ReplicaManager(val config: KafkaConfig,
     inklessFetchOffsetHandler.foreach(_.close())
     inklessRetentionEnforcer.foreach(_.close())
     inklessFileCleaner.foreach(_.close())
+    inklessTopicPurger.foreach(_.close())
     inklessDeleteRecordsInterceptor.foreach(_.close())
     inklessSharedState.foreach(_.close())
     info("Shut down completely")

diff --git a/core/src/test/java/kafka/server/InklessDisklessTopicDeleteTest.java b/core/src/test/java/kafka/server/InklessDisklessTopicDeleteTest.java
diff --git a/docs/inkless/FEATURES.md b/docs/inkless/FEATURES.md
@@ -47,6 +47,7 @@ If not specified above, features are untested and assumed to be inoperable.
     - when `diskless.managed.rf.enable=true`: any valid RF is accepted — RF=-1 resolves to `default.replication.factor`, explicit RF values (1, 2, 3, ...) are accepted, and placement uses standard rack-aware assignment;
     - manual replica assignments are accepted only when `diskless.managed.rf.enable=true` (rejected in legacy mode).
 - `DELETE_TOPICS`
+    - the response returns after the controller stamps the diskless logs deleted. Brokers drain batches in the background. Objects stay until that drain finishes and FileCleaner passes `file.cleaner.retention.period.ms`. Size `topic.purger.max.batches.per.cycle` and `topic.purger.interval.ms` for the drain rate you want.
 - `DELETE_RECORDS`
 - `OFFSET_FOR_LEADER_EPOCH`
 - `DESCRIBE_CONFIGS`

diff --git a/docs/inkless/configs.rst b/docs/inkless/configs.rst
@@ -274,6 +274,22 @@ Under ``inkless.``
   * Valid Values: [0,...]
   * Importance: low
 
+``topic.purger.interval.ms``
+  The interval with which to drain batches from soft-deleted diskless logs. DELETE_TOPICS returns after the controller stamps the logs. This interval and topic.purger.max.batches.per.cycle set how fast brokers reclaim those rows. Every broker runs its own purger.
+
+  * Type: int
+  * Default: 60000 (1 minute)
+  * Valid Values: [1,...]
+  * Importance: low
+
+``topic.purger.max.batches.per.cycle``
+  The maximum number of batches a single topic purge cycle may delete, per broker. The same bound applies to empty soft-deleted logs dropped in that cycle. The per-broker drain rate is at most this value divided by topic.purger.interval.ms, and every broker runs its own purger. With the defaults that is 20000 batches per broker per minute. A deleted topic's objects stay until the purger marks the files and FileCleaner passes file.cleaner.retention.period.ms. Each cycle is one transaction. 0 drains every remaining batch in that transaction and holds back autovacuum on the control-plane database until it commits. Use 0 only in tests or a maintenance window.
+
+  * Type: int
+  * Default: 20000
+  * Valid Values: [0,...]
+  * Importance: low
+
 
 
 -----------------

diff --git a/docs/inkless/metrics.rst b/docs/inkless/metrics.rst
@@ -249,6 +249,27 @@ LastSuccessfulFileCleanupAgeMs   Milliseconds since the last file cleaning cycle
 ===============================  ==================================================================================================================================================================
 
 
+TopicPurger metrics
+==================================
+
+io.aiven.inkless.delete:type=TopicPurger
+----------------------------------------
+
+==============================  ============================================================================================================================================================================================================================================================================
+Attribute name                  Description                                                                                                                                                                                                                                                                 
+==============================  ============================================================================================================================================================================================================================================================================
+LastSuccessfulTopicPurgeAgeMs   Milliseconds since the last topic purge cycle completed without error, including cycles that found nothing to purge; -1 if no cycle has completed since startup. This stays fresh while a backlog remains, so pair it with TopicPurgerWorkRemain and TopicPurgerBatchesRate.
+TopicPurgerBatchesRate          Total number of batches deleted from soft-deleted logs                                                                                                                                                                                                                      
+TopicPurgerCycleSaturatedRate   Total number of topic purge cycles that hit topic.purger.max.batches.per.cycle, leaving work for the next cycle                                                                                                                                                             
+TopicPurgerErrorRate            Total number of topic purge errors                                                                                                                                                                                                                                          
+TopicPurgerFilesMarkedRate      Total number of files marked for deletion by topic purge                                                                                                                                                                                                                    
+TopicPurgerLogsRate             Total number of soft-deleted logs fully purged                                                                                                                                                                                                                              
+TopicPurgerRate                 Total number of topic purge cycles started                                                                                                                                                                                                                                  
+TopicPurgerTotalTime            Total time spent on a topic purge cycle in milliseconds                                                                                                                                                                                                                     
+TopicPurgerWorkRemain           1 if the last successful topic purge cycle left soft-deleted logs, including rows another broker holds; 0 otherwise                                                                                                                                                         
+==============================  ============================================================================================================================================================================================================================================================================
+
+
 RetentionEnforcer metrics
 ==================================
 
@@ -348,6 +369,9 @@ InitDisklessLogQueryTime                           Time spent executing the Init
 ListOffsetsLastSuccessfulQueryAgeMs                Milliseconds since the last successful ListOffsets query completed; -1 if no query has succeeded since startup             
 ListOffsetsQueryRate                               Total number of ListOffsets queries executed                                                                               
 ListOffsetsQueryTime                               Time spent executing the ListOffsets query in milliseconds                                                                 
+PurgeDeletedLogsLastSuccessfulQueryAgeMs           Milliseconds since the last successful PurgeDeletedLogs query completed; -1 if no query has succeeded since startup        
+PurgeDeletedLogsQueryRate                          Total number of PurgeDeletedLogs queries executed                                                                          
+PurgeDeletedLogsQueryTime                          Time spent executing the PurgeDeletedLogs query in milliseconds                                                            
 RepairDisklessLogLastSuccessfulQueryAgeMs          Milliseconds since the last successful RepairDisklessLog query completed; -1 if no query has succeeded since startup       
 RepairDisklessLogQueryRate                         Total number of RepairDisklessLog queries executed                                                                         
 RepairDisklessLogQueryTime                         Time spent executing the RepairDisklessLog query in milliseconds                                                           

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
@@ -162,6 +162,24 @@ public class InklessConfig extends AbstractConfig {
         "unbounded.";
     private static final int FILE_CLEANER_MAX_FILES_PER_CYCLE_DEFAULT = 20_000;
 
+    public static final String TOPIC_PURGER_INTERVAL_MS_CONFIG = "topic.purger.interval.ms";
+    private static final String TOPIC_PURGER_INTERVAL_MS_DOC = "The interval with which to drain batches from "
+        + "soft-deleted diskless logs. DELETE_TOPICS returns after the controller stamps the logs. This "
+        + "interval and topic.purger.max.batches.per.cycle set how fast brokers reclaim those rows. Every "
+        + "broker runs its own purger.";
+    private static final int TOPIC_PURGER_INTERVAL_MS_DEFAULT = 60 * 1000;  // 1 minute
+
+    public static final String TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_CONFIG = "topic.purger.max.batches.per.cycle";
+    private static final String TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_DOC = "The maximum number of batches a single "
+        + "topic purge cycle may delete, per broker. The same bound applies to empty soft-deleted logs dropped "
+        + "in that cycle. The per-broker drain rate is at most this value divided by "
+        + "topic.purger.interval.ms, and every broker runs its own purger. With the defaults that is 20000 "
+        + "batches per broker per minute. A deleted topic's objects stay until the purger marks the files and "
+        + "FileCleaner passes file.cleaner.retention.period.ms. Each cycle is one transaction. 0 drains every "
+        + "remaining batch in that transaction and holds back autovacuum on the control-plane database until "
+        + "it commits. Use 0 only in tests or a maintenance window.";
+    private static final int TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_DEFAULT = 20_000;
+
 
     public static final String PRODUCE_UPLOAD_THREAD_POOL_SIZE_CONFIG = "produce.upload.thread.pool.size";
     private static final String PRODUCE_UPLOAD_THREAD_POOL_SIZE_DOC = "Thread pool size to concurrently upload files to remote storage";
@@ -398,6 +416,24 @@ public static ConfigDef configDef() {
             FILE_CLEANER_MAX_FILES_PER_CYCLE_DOC
         );
 
+        configDef.define(
+            TOPIC_PURGER_INTERVAL_MS_CONFIG,
+            ConfigDef.Type.INT,
+            TOPIC_PURGER_INTERVAL_MS_DEFAULT,
+            ConfigDef.Range.atLeast(1),
+            ConfigDef.Importance.LOW,
+            TOPIC_PURGER_INTERVAL_MS_DOC
+        );
+
+        configDef.define(
+            TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_CONFIG,
+            ConfigDef.Type.INT,
+            TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_DEFAULT,
+            ConfigDef.Range.atLeast(0),  // 0 = unbounded
+            ConfigDef.Importance.LOW,
+            TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_DOC
+        );
+
         configDef.define(
             CROSS_TIER_LOG_START_REPORT_INTERVAL_MS_CONFIG,
             ConfigDef.Type.INT,
@@ -758,6 +794,14 @@ public int fileCleanerMaxFilesPerCycle() {
         return getInt(FILE_CLEANER_MAX_FILES_PER_CYCLE_CONFIG);
     }
 
+    public Duration topicPurgerInterval() {
+        return Duration.ofMillis(getInt(TOPIC_PURGER_INTERVAL_MS_CONFIG));
+    }
+
+    public int topicPurgerMaxBatchesPerCycle() {
+        return getInt(TOPIC_PURGER_MAX_BATCHES_PER_CYCLE_CONFIG);
+    }
+
     public Duration crossTierLogStartReportInterval() {
         return Duration.ofMillis(getInt(CROSS_TIER_LOG_START_REPORT_INTERVAL_MS_CONFIG));
     }

diff --git a/storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlane.java b/storage/inkless/src/main/java/io/aiven/inkless/control_plane/ControlPlane.java
@@ -82,6 +82,16 @@ List<FindBatchResponse> findBatches(
 
     void deleteTopics(Set<Uuid> topicIds);
 
+    /**
+     * Drain batches belonging to soft-deleted logs, mark newly empty files, and drop logs that have
+     * no remaining batches.
+     *
+     * @param maxBatches maximum batches to delete this call, and the cap on empty
+     *                   soft-deleted logs dropped this call; {@code 0} means unbounded
+     * @return how much work this call did
+     */
+    PurgeDeletedLogsResponse purgeDeletedLogs(int maxBatches);
+
     List<EnforceRetentionResponse> enforceRetention(List<EnforceRetentionRequest> requests, int maxBatchesPerRequest);
 
     /**