fix(s3stream): fix compaction exit on empty stream metadata list (#2095)

Signed-off-by: Shichao Nie <[email protected]>

Author: SCNieh

s3stream/src/main/java/com/automq/stream/s3/compact/CompactionManager.java

@@ -231,10 +231,6 @@ public CompletableFuture<Void> compact() {
                 return CompletableFuture.failedFuture(e);
             }
             return this.streamManager.getStreams(new ArrayList<>(streamIds)).thenAcceptAsync(streamMetadataList -> {
-                if (streamMetadataList.isEmpty()) {
-                    logger.info("No stream metadata found for stream set objects");
-                    return;
-                }
                 filterInvalidStreamDataBlocks(streamMetadataList);
                 this.compact(streamMetadataList, objectMetadataList);
             }, compactThreadPool);
@@ -425,10 +421,6 @@ public CompletableFuture<Void> forceSplitAll() {
             List<Long> streamIds = streamDataBlockMap.values().stream().flatMap(Collection::stream)
                 .map(StreamDataBlock::getStreamId).distinct().collect(Collectors.toList());
             this.streamManager.getStreams(streamIds).thenAcceptAsync(streamMetadataList -> {
-                if (streamMetadataList.isEmpty()) {
-                    logger.info("No stream metadata found for stream set objects");
-                    return;
-                }
                 filterInvalidStreamDataBlocks(streamMetadataList);
                 forceSplitObjects(streamMetadataList, objectMetadataList);
                 cf.complete(null);

s3stream/src/test/java/com/automq/stream/s3/compact/CompactionManagerTest.java

@@ -201,6 +201,72 @@ public void testForceSplitWithOutDatedObject() {
         Assertions.assertTrue(request.getStreamRanges().isEmpty());
     }
 
+
+    @Test
+    public void testForceSplitWithNonExistStream() {
+        List<StreamMetadata> streamMetadataList = this.streamManager.getStreams(Collections.emptyList()).join();
+        streamMetadataList = streamMetadataList.stream().filter(s -> s.streamId() != STREAM_0).collect(Collectors.toList());
+        List<S3ObjectMetadata> s3ObjectMetadata = this.objectManager.getServerObjects().join();
+        when(config.streamSetObjectCompactionForceSplitPeriod()).thenReturn(0);
+        compactionManager = new CompactionManager(config, objectManager, streamManager, objectStorage);
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(0)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        CommitStreamSetObjectRequest request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(0));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_0), request.getCompactedObjectIds());
+        Assertions.assertEquals(2, request.getStreamObjects().size());
+        Assertions.assertTrue(checkDataIntegrity(streamMetadataList, Collections.singletonList(s3ObjectMetadata.get(0)), request));
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(1)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(1));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_1), request.getCompactedObjectIds());
+        Assertions.assertEquals(1, request.getStreamObjects().size());
+        Assertions.assertTrue(checkDataIntegrity(streamMetadataList, Collections.singletonList(s3ObjectMetadata.get(1)), request));
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(2)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(2));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_2), request.getCompactedObjectIds());
+        Assertions.assertEquals(2, request.getStreamObjects().size());
+        Assertions.assertTrue(checkDataIntegrity(streamMetadataList, Collections.singletonList(s3ObjectMetadata.get(2)), request));
+    }
+
+    @Test
+    public void testForceSplitWithEmptyStreamList() {
+        List<StreamMetadata> streamMetadataList = Collections.emptyList();
+        List<S3ObjectMetadata> s3ObjectMetadata = this.objectManager.getServerObjects().join();
+        when(config.streamSetObjectCompactionForceSplitPeriod()).thenReturn(0);
+        compactionManager = new CompactionManager(config, objectManager, streamManager, objectStorage);
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(0)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        CommitStreamSetObjectRequest request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(0));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_0), request.getCompactedObjectIds());
+        Assertions.assertTrue(request.getStreamObjects().isEmpty());
+        Assertions.assertTrue(request.getStreamRanges().isEmpty());
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(1)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(1));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_1), request.getCompactedObjectIds());
+        Assertions.assertTrue(request.getStreamObjects().isEmpty());
+        Assertions.assertTrue(request.getStreamRanges().isEmpty());
+
+        compactionManager.updateStreamDataBlockMap(List.of(s3ObjectMetadata.get(2)));
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        request = compactionManager.buildSplitRequest(streamMetadataList, s3ObjectMetadata.get(2));
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertEquals(List.of(OBJECT_2), request.getCompactedObjectIds());
+        Assertions.assertTrue(request.getStreamObjects().isEmpty());
+        Assertions.assertTrue(request.getStreamRanges().isEmpty());
+    }
+
     @Test
     public void testForceSplitWithException() {
         S3AsyncClient s3AsyncClient = Mockito.mock(S3AsyncClient.class);
@@ -465,6 +531,21 @@ public void testCompactWithNonExistStream() {
         Assertions.assertTrue(checkDataIntegrity(streamMetadataList, S3_WAL_OBJECT_METADATA_LIST, request));
     }
 
+    @Test
+    public void testCompactWithEmptyStream() {
+        compactionManager = new CompactionManager(config, objectManager, streamManager, objectStorage);
+        List<StreamMetadata> streamMetadataList = Collections.emptyList();
+        compactionManager.updateStreamDataBlockMap(S3_WAL_OBJECT_METADATA_LIST);
+        compactionManager.filterInvalidStreamDataBlocks(streamMetadataList);
+        CommitStreamSetObjectRequest request = compactionManager.buildCompactRequest(streamMetadataList, S3_WAL_OBJECT_METADATA_LIST);
+
+        Assertions.assertEquals(-1, request.getObjectId());
+        Assertions.assertTrue(request.getStreamObjects().isEmpty());
+        Assertions.assertTrue(request.getStreamRanges().isEmpty());
+        assertEquals(List.of(OBJECT_0, OBJECT_1, OBJECT_2), request.getCompactedObjectIds());
+        Assertions.assertTrue(checkDataIntegrity(streamMetadataList, S3_WAL_OBJECT_METADATA_LIST, request));
+    }
+
     @Test
     public void testCompactNoneExistObjects() {
         when(config.streamSetObjectCompactionStreamSplitSize()).thenReturn(100L);