HDDS-8882. Manage status of DeleteBlocksCommand in SCM to avoid sending duplicates to Datanode (apache#4988)

(cherry picked from commit 88e18e3)

Author: xichen01

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/CommandStatusReportPublisher.java

@@ -78,9 +78,9 @@ protected CommandStatusReportsProto getReport() {
       // If status is still pending then don't remove it from map as
       // CommandHandler will change its status when it works on this command.
       if (!cmdStatus.getStatus().equals(Status.PENDING)) {
-        builder.addCmdStatus(cmdStatus.getProtoBufMessage());
         map.remove(key);
       }
+      builder.addCmdStatus(cmdStatus.getProtoBufMessage());
     });
     return builder.getCmdStatusCount() > 0 ? builder.build() : null;
   }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/report/TestReportPublisher.java

@@ -173,8 +173,8 @@ public void testCommandStatusPublisher() throws InterruptedException {
         .build();
     cmdStatusMap.put(obj1.getCmdId(), obj1);
     cmdStatusMap.put(obj2.getCmdId(), obj2);
-    // We are not sending the commands whose status is PENDING.
-    Assertions.assertEquals(1,
+    // We will sending the commands whose status is PENDING and EXECUTED
+    Assertions.assertEquals(2,
         ((CommandStatusReportPublisher) publisher).getReport()
             .getCmdStatusCount(),
         "Should publish report with 2 status objects");

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java

@@ -19,12 +19,10 @@
 
 import java.util.Set;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.proto
-    .StorageContainerDatanodeProtocolProtos.ContainerBlocksDeletionACKProto
-    .DeleteBlockTransactionResult;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -88,14 +86,33 @@ void incrementCount(List<Long> txIDs)
   int resetCount(List<Long> txIDs) throws IOException;
 
   /**
-   * Commits a transaction means to delete all footprints of a transaction
-   * from the log. This method doesn't guarantee all transactions can be
-   * successfully deleted, it tolerate failures and tries best efforts to.
-   *  @param transactionResults - delete block transaction results.
-   * @param dnID - ID of datanode which acknowledges the delete block command.
+   * Records the creation of a transaction for a DataNode.
+   *
+   * @param dnId The identifier of the DataNode.
+   * @param scmCmdId The ID of the SCM command.
+   * @param dnTxSet Set of transaction IDs for the DataNode.
+   */
+  void recordTransactionCreated(
+      UUID dnId, long scmCmdId, Set<Long> dnTxSet);
+
+  /**
+   * Handles the cleanup process when a DataNode is reported dead. This method
+   * is responsible for updating or cleaning up the transaction records
+   * associated with the dead DataNode.
+   *
+   * @param dnId The identifier of the dead DataNode.
+   */
+  void onDatanodeDead(UUID dnId);
+
+  /**
+   * Records the event of sending a block deletion command to a DataNode. This
+   * method is called when a command is successfully dispatched to a DataNode,
+   * and it helps in tracking the status of the command.
+   *
+   * @param dnId Details of the DataNode.
+   * @param scmCommand The block deletion command sent.
    */
-  void commitTransactions(List<DeleteBlockTransactionResult> transactionResults,
-      UUID dnID);
+  void onSent(DatanodeDetails dnId, SCMCommand<?> scmCommand);
 
   /**
    * Creates block deletion transactions for a set of containers,