HDDS-8869. Make DN DeleteBlocksCommandHandler wait for the lock can timeout.

hadoop-hdds/container-service/dev-support/findbugsExcludeFile.xml

@@ -65,6 +65,10 @@
     <Class name="org.apache.hadoop.ozone.container.common.statemachine.TestStateContext" />
     <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
   </Match>
+  <Match>
+    <Class name="org.apache.hadoop.ozone.container.common.statemachine.commandhandler.TestDeleteBlocksCommandHandler" />
+    <Bug pattern="URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD" />
+  </Match>
   <Match>
     <Class name="org.apache.hadoop.ozone.container.common.volume.TestVolumeSet" />
     <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/BlockDeletingServiceMetrics.java

@@ -74,6 +74,10 @@ public final class BlockDeletingServiceMetrics {
   @Metric(about = "The total number of Container chosen to be deleted.")
   private MutableGaugeLong totalContainerChosenCount;
 
+  @Metric(about = "The total number of transactions which failed due" +
+      " to container lock wait timeout.")
+  private MutableGaugeLong totalLockTimeoutTransactionCount;
+
   private BlockDeletingServiceMetrics() {
   }
 
@@ -140,6 +144,10 @@ public void setTotalPendingBlockCount(long count) {
     this.totalPendingBlockCount.set(count);
   }
 
+  public void incrTotalLockTimeoutTransactionCount() {
+    totalLockTimeoutTransactionCount.incr();
+  }
+
   public long getSuccessCount() {
     return successCount.value();
   }
@@ -172,6 +180,10 @@ public long getTotalContainerChosenCount() {
     return totalContainerChosenCount.value();
   }
 
+  public long getTotalLockTimeoutTransactionCount() {
+    return totalLockTimeoutTransactionCount.value();
+  }
+
   @Override
   public String toString() {
     StringBuffer buffer = new StringBuffer();
@@ -195,7 +207,9 @@ public String toString() {
         .append("receivedBlockCount = "
             + receivedBlockCount.value()).append("\t")
         .append("markedBlockCount = "
-            + markedBlockCount.value()).append("\t");
+            + markedBlockCount.value()).append("\t")
+        .append("totalLockTimeoutTransactionCount = "
+            + totalLockTimeoutTransactionCount.value()).append("\t");
     return buffer.toString();
   }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeConfiguration.java

@@ -188,6 +188,23 @@ public class DatanodeConfiguration {
   private long recoveringContainerScrubInterval =
       Duration.ofMinutes(10).toMillis();
 
+  /**
+   * The maximum time to wait for acquiring the container lock when processing
+   * a delete block transaction.
+   * If a timeout occurs while attempting to get the lock, the delete block
+   * transaction won't be immediately discarded. Instead, it will be retried
+   * after all the current delete block transactions have been processed.
+   */
+  @Config(key = "block.delete.max.lock.wait.timeout",
+      defaultValue = "100ms",
+      type = ConfigType.TIME,
+      tags = { DATANODE, ConfigTag.DELETION},
+      description = "Timeout for the thread used to process the delete" +
+          " block command to wait for the container lock."
+  )
+  private long blockDeleteMaxLockWaitTimeoutMs =
+      Duration.ofMillis(100).toMillis();
+
   public Duration getBlockDeletionInterval() {
     return Duration.ofMillis(blockDeletionInterval);
   }
@@ -573,6 +590,10 @@ public int getBlockDeleteQueueLimit() {
     return blockDeleteQueueLimit;
   }
 
+  public long getBlockDeleteMaxLockWaitTimeoutMs() {
+    return blockDeleteMaxLockWaitTimeoutMs;
+  }
+
   public void setBlockDeleteQueueLimit(int queueLimit) {
     this.blockDeleteQueueLimit = queueLimit;
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -225,8 +225,7 @@ public DatanodeStateMachine(DatanodeDetails datanodeDetails,
     commandDispatcher = CommandDispatcher.newBuilder()
         .addHandler(new CloseContainerCommandHandler())
         .addHandler(new DeleteBlocksCommandHandler(getContainer(),
-            conf, dnConf.getBlockDeleteThreads(),
-            dnConf.getBlockDeleteQueueLimit()))
+            conf, dnConf))
         .addHandler(new ReplicateContainerCommandHandler(conf, supervisor,
             pullReplicatorWithMetrics, pushReplicatorWithMetrics))
         .addHandler(reconstructECContainersCommandHandler)