feat(issues1842): cleanup metastream kv after deleting topic (#1843)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -2157,7 +2157,7 @@ private QuorumController(
         this.streamControlManager = new StreamControlManager(this, snapshotRegistry, logContext,
                 this.s3ObjectControlManager, clusterControl, featureControl, replicationControl);
         this.kvControlManager = new KVControlManager(snapshotRegistry, logContext);
-        this.topicDeletionManager = new TopicDeletionManager(snapshotRegistry, this, streamControlManager);
+        this.topicDeletionManager = new TopicDeletionManager(snapshotRegistry, this, streamControlManager, kvControlManager);
         this.nodeControlManager = new NodeControlManager(snapshotRegistry, new DefaultNodeRuntimeInfoGetter(clusterControl, streamControlManager));
         this.extension = extension.apply(this);
 

metadata/src/main/java/org/apache/kafka/controller/stream/TopicDeletionManager.java

@@ -11,9 +11,11 @@
 
 package org.apache.kafka.controller.stream;
 
+import com.automq.stream.s3.metadata.ObjectUtils;
 import com.automq.stream.s3.metadata.StreamState;
 import com.automq.stream.utils.Threads;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
@@ -50,17 +52,20 @@ public class TopicDeletionManager {
 
     private final Controller quorumController;
     private final StreamControlManager streamControlManager;
+    private final KVControlManager kvControlManager;
 
     private CompletableFuture<Void> lastDeletion = CompletableFuture.completedFuture(null);
 
     public TopicDeletionManager(
         SnapshotRegistry registry,
         Controller quorumController,
-        StreamControlManager streamControlManager
+        StreamControlManager streamControlManager,
+        KVControlManager kvControlManager
     ) {
         this.waitingCleanupTopics = new TimelineHashMap<>(registry, 0);
         this.quorumController = quorumController;
         this.streamControlManager = streamControlManager;
+        this.kvControlManager = kvControlManager;
         Threads.COMMON_SCHEDULER.scheduleWithFixedDelay(this::cleanupDeletedTopics, 5, 5, TimeUnit.SECONDS);
     }
 
@@ -98,7 +103,16 @@ public ControllerResult<Void> cleanupDeletedTopics0() {
             }
         }
         if (streamsToDelete.isEmpty()) {
-            List<ApiMessageAndVersion> records = List.of(new ApiMessageAndVersion(new RemoveKVRecord().setKeys(List.of(TopicDeletion.TOPIC_DELETION_PREFIX + topicId)), (short) 0));
+            List<ApiMessageAndVersion> records = new ArrayList<>();
+            List<String> metaStreamKvList = new ArrayList<>();
+            String metaStreamKvPrefix = ObjectUtils.genMetaStreamKvPrefix(topicId);
+            kvControlManager.kv().forEach((k, v) -> {
+                if (k.startsWith(metaStreamKvPrefix)) {
+                    metaStreamKvList.add(k);
+                }
+            });
+            records.add(new ApiMessageAndVersion(new RemoveKVRecord().setKeys(metaStreamKvList), (short) 0));
+            records.add(new ApiMessageAndVersion(new RemoveKVRecord().setKeys(List.of(TopicDeletion.TOPIC_DELETION_PREFIX + topicId)), (short) 0));
             LOGGER.info("Topic clean up completed for topic {}", topicId);
             if (waitingCleanupTopics.size(lastStableOffset) > 1) {
                 // there are more topics to clean up, fast trigger next round cleanup

metadata/src/test/java/org/apache/kafka/controller/stream/TopicDeletionManagerTest.java

@@ -11,7 +11,9 @@
 
 package org.apache.kafka.controller.stream;
 
+import com.automq.stream.s3.metadata.ObjectUtils;
 import com.automq.stream.s3.metadata.StreamState;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.HashMap;
@@ -51,8 +53,13 @@ public class TopicDeletionManagerTest {
     final Uuid topicId = new Uuid(1, 2);
     SnapshotRegistry registry;
     Controller quorumController;
+
     TimelineHashMap<Long, StreamRuntimeMetadata> streams;
     StreamControlManager streamControlManager;
+
+    TimelineHashMap<String, ByteBuffer> kvs;
+    KVControlManager kvControlManager;
+
     TopicDeletionManager topicDeletionManager;
 
     @BeforeEach
@@ -63,7 +70,12 @@ public void setup() {
         streams = new TimelineHashMap<>(registry, 0);
         streamControlManager = mock(StreamControlManager.class);
         when(streamControlManager.streamsMetadata()).thenReturn(streams);
-        topicDeletionManager = new TopicDeletionManager(registry, quorumController, streamControlManager);
+
+        kvs = new TimelineHashMap<>(registry, 0);
+        kvControlManager = mock(KVControlManager.class);
+        when(kvControlManager.kv()).thenReturn(kvs);
+
+        topicDeletionManager = new TopicDeletionManager(registry, quorumController, streamControlManager, kvControlManager);
     }
 
     @Test
@@ -75,6 +87,13 @@ public void testCleanup() {
         when(quorumController.isActive()).thenReturn(true);
         when(quorumController.appendWriteEvent(any(), any(), any())).thenReturn(CompletableFuture.completedFuture(null));
 
+        List<String> metadataKvList = List.of(
+            ObjectUtils.genMetaStreamKvPrefix(topicId.toString()) + 1,
+            ObjectUtils.genMetaStreamKvPrefix(topicId.toString()) + 2
+        );
+        metadataKvList.forEach(str -> kvs.put(str, ByteBuffer.wrap(new byte[0])));
+        kvs.put(ObjectUtils.genMetaStreamKvPrefix(topicId + "others") + 1, ByteBuffer.wrap(new byte[0]));
+
         streams.put(1L, new StreamRuntimeMetadata(1L, 0, 0, 0,
             StreamState.CLOSED, Collections.emptyMap(), registry));
         Map<String, String> tags = new HashMap<>();
@@ -122,7 +141,11 @@ public void testCleanup() {
         registry.getOrCreateSnapshot(4L);
         when(quorumController.lastStableOffset()).thenReturn(4L);
 
-        replay(topicDeletionManager.cleanupDeletedTopics0(), topicDeletionManager);
+        ControllerResult<Void> rst = topicDeletionManager.cleanupDeletedTopics0();
+        assertEquals(2, rst.records().size());
+        assertEquals(metadataKvList.stream().sorted().collect(Collectors.toList()), ((RemoveKVRecord) rst.records().get(0).message()).keys().stream().sorted().collect(Collectors.toList()));
+        assertEquals(List.of(TopicDeletion.TOPIC_DELETION_PREFIX + topicId), ((RemoveKVRecord) rst.records().get(1).message()).keys());
+        replay(rst, topicDeletionManager);
         verify(quorumController, times(2)).deleteStreams(any(), any());
         assertEquals(0, topicDeletionManager.waitingCleanupTopics.size());
     }

s3stream/src/main/java/com/automq/stream/s3/metadata/ObjectUtils.java

@@ -28,8 +28,8 @@ public static void setNamespace(String namespace) {
         ObjectUtils.namespace = namespace;
     }
 
-    public static void main(String[] args) {
-        System.out.printf("%s%n", genKey(0, 11154));
+    public static String genMetaStreamKvPrefix(String topic) {
+        return namespace + "/" + topic + "/";
     }
 
     public static String genIndexKey(int version, long nodeId) {