feat(metadata): cleanup object wait last round cleanup record replay … (

#1476) feat(metadata): cleanup object wait last round cleanup record replay completed Signed-off-by: Robin Han <[email protected]>
AutoMQ · Jun 27, 2024 · 05e530e · 05e530e
1 parent 26ffc02
commit 05e530e
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 2 deletions.
diff --git a/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java
@@ -376,18 +376,20 @@ private CompletableFuture<Void> clean0(List<String> objectKeys) {
                     log.error("Failed to delete the S3Object from S3, objectKeys: {}",
                         String.join(",", objectKeys), e);
                     return null;
-                }).thenAccept(resp -> {
+                }).thenCompose(resp -> {
                     if (resp != null && !resp.isEmpty()) {
                         List<Long> deletedObjectIds = resp.stream().map(key -> ObjectUtils.parseObjectId(0, key)).collect(Collectors.toList());
                         // notify the controller an objects deletion event to drive the removal of the objects
                         ControllerRequestContext ctx = new ControllerRequestContext(
                             null, null, OptionalLong.empty());
-                        quorumController.notifyS3ObjectDeleted(ctx, deletedObjectIds).whenComplete((ignore, exp) -> {
+                        return quorumController.notifyS3ObjectDeleted(ctx, deletedObjectIds).whenComplete((ignore, exp) -> {
                             if (exp != null) {
                                 log.error("Failed to notify the controller the S3Object deletion event, objectIds: {}",
                                     Arrays.toString(deletedObjectIds.toArray()), exp);
                             }
                         });
+                    } else {
+                        return CompletableFuture.completedFuture(null);
                     }
                 });
         }

diff --git a/metadata/src/test/java/org/apache/kafka/controller/S3ObjectControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/S3ObjectControlManagerTest.java
@@ -50,6 +50,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.when;
 
 @Timeout(40)
 @Tag("S3Unit")
@@ -71,6 +72,7 @@ public class S3ObjectControlManagerTest {
     @BeforeEach
     public void setUp() {
         controller = Mockito.mock(QuorumController.class);
+        when(controller.notifyS3ObjectDeleted(any(), anyList())).thenReturn(CompletableFuture.completedFuture(null));
         operator = Mockito.mock(S3Operator.class);
         Mockito.when(operator.delete(anyList())).then(inv -> {
             List<String> objectKeys = inv.getArgument(0);