HDDS-14418. Skip sending block deletion command to SCM for empty files.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/KeyDeletingService.java

@@ -139,21 +139,49 @@ Pair<Pair<Integer, Long>, Boolean> processKeyDeletes(Map<String, PurgedKey> keyB
       String snapTableKey, UUID expectedPreviousSnapshotId) throws IOException {
     long startTime = Time.monotonicNow();
     Pair<Pair<Integer, Long>, Boolean> purgeResult = Pair.of(Pair.of(0, 0L), false);
+
+    // Filter out empty files (files with no blocks) before sending to SCM
+    Map<String, PurgedKey> nonEmptyKeyBlocksList = keyBlocksList.entrySet().stream()
+        .filter(entry -> entry.getValue().getBlockGroup() != null && 
+                         entry.getValue().getBlockGroup().getDeletedBlocks() != null &&
+                         !entry.getValue().getBlockGroup().getDeletedBlocks().isEmpty())
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Send {} key(s) to SCM: {}",
-          keyBlocksList.size(), keyBlocksList);
+      LOG.debug("Send {} key(s) to SCM (filtered {} empty keys): {}",
+          nonEmptyKeyBlocksList.size(), keyBlocksList.size() - nonEmptyKeyBlocksList.size(), nonEmptyKeyBlocksList);
     } else if (LOG.isInfoEnabled()) {
       int logSize = 10;
-      if (keyBlocksList.size() < logSize) {
-        logSize = keyBlocksList.size();
+      if (nonEmptyKeyBlocksList.size() < logSize) {
+        logSize = nonEmptyKeyBlocksList.size();
       }
       LOG.info("Send {} key(s) to SCM, first {} keys: {}",
           keyBlocksList.size(), logSize, keyBlocksList.entrySet().stream().limit(logSize)
               .map(Map.Entry::getValue).collect(Collectors.toSet()));
     }
-    List<DeleteBlockGroupResult> blockDeletionResults =
-        scmClient.deleteKeyBlocks(keyBlocksList.values().stream()
-            .map(PurgedKey::getBlockGroup).collect(Collectors.toList()));
+    List<DeleteBlockGroupResult> blockDeletionResults;
+    if (nonEmptyKeyBlocksList.isEmpty()) {
+      // Skip SCM call if all files are empty
+      blockDeletionResults = new ArrayList<>();
+      LOG.info("Skipping SCM call as all {} keys are empty", keyBlocksList.size());
+    } else {
+      blockDeletionResults =
+          scmClient.deleteKeyBlocks(nonEmptyKeyBlocksList.values().stream()
+              .map(PurgedKey::getBlockGroup).collect(Collectors.toList()));
+    }
+
+    // Add successful results for empty files (no need to send to SCM)
+    Map<String, PurgedKey> emptyKeyBlocksList = keyBlocksList.entrySet().stream()
+        .filter(entry -> !nonEmptyKeyBlocksList.containsKey(entry.getKey()))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    for (PurgedKey emptyKey : emptyKeyBlocksList.values()) {
+      // Create a successful result for empty files
+      DeleteBlockGroupResult emptyFileResult = new DeleteBlockGroupResult(
+          emptyKey.getBlockGroup().getGroupID(), new ArrayList<>());
+      blockDeletionResults.add(emptyFileResult);
+    }
+
     LOG.info("{} BlockGroup deletion are acked by SCM in {} ms",
         keyBlocksList.size(), Time.monotonicNow() - startTime);
     if (blockDeletionResults != null) {

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyDeletingService.java

@@ -39,6 +39,7 @@
 import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -246,6 +247,32 @@ void checkIfDeleteServiceIsDeletingKeys()
       }
     }
 
+    /**
+     * Test that verifies zero-sized keys (keys with no blocks) are not sent to SCM.
+     * The KeyDeletingService should filter out empty keys before calling SCM.
+     */
+    @Test
+    void checkIfDeleteServiceIsDeletingZeroSizedKeys()
+        throws IOException, TimeoutException, InterruptedException {
+      // Spy on the SCM client to verify it's not called for empty keys
+      ScmBlockLocationTestingClient scmClientSpy = Mockito.spy(scmBlockTestingClient);
+      // Create a KeyDeletingService with the spied client
+      KeyDeletingService testService = new KeyDeletingService(
+          om, scmClientSpy, 100, 10000, conf, 10, false);
+      // Create a BlockGroup with empty deleted blocks list (zero-sized key)
+      BlockGroup blockGroup = BlockGroup.newBuilder().setKeyName("key1/1")
+          .addAllDeletedBlocks(new ArrayList<>()).build();
+      Map<String, PurgedKey> blockGroups = Collections.singletonMap(
+          blockGroup.getGroupID(), 
+          new PurgedKey("vol", "buck", 1, blockGroup, "key1", 0, true));
+      // Process the key deletion
+      testService.processKeyDeletes(blockGroups, new HashMap<>(), new ArrayList<>(), null, null);
+      // Verify that SCM's deleteKeyBlocks was never called (empty keys are filtered out)
+      verify(scmClientSpy, never()).deleteKeyBlocks(any());
+      // Cleanup
+      testService.shutdown();
+    }
+
     @Test
     void checkDeletionForKeysWithMultipleVersions() throws Exception {
       final long initialDeletedCount = getDeletedKeyCount();