HDFS-17372. CommandProcessingThread#queue should use LinkedBlockingDeque to prevent high priority command blocked by low priority command.

Author: hfutatzhanghb

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java

@@ -178,7 +178,7 @@ public HeartbeatResponse sendHeartbeat(DatanodeRegistration registration,
     }
     return new HeartbeatResponse(cmds, PBHelper.convert(resp.getHaStatus()),
         rollingUpdateStatus, resp.getFullBlockReportLeaseId(),
-        resp.getIsSlownode());
+        resp.getIsSlownode(), resp.getIsCommandHighPriority());
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java

@@ -1657,13 +1657,15 @@ public void setBlockToken(final LocatedBlock b,
     }
   }
 
-  void addKeyUpdateCommand(final List<DatanodeCommand> cmds,
+  boolean addKeyUpdateCommand(final List<DatanodeCommand> cmds,
       final DatanodeDescriptor nodeinfo) {
     // check access key update
     if (isBlockTokenEnabled() && nodeinfo.needKeyUpdate()) {
       cmds.add(new KeyUpdateCommand(blockTokenSecretManager.exportKeys()));
       nodeinfo.setNeedKeyUpdate(false);
+      return true;
     }
+    return false;
   }
 
   public DataEncryptionKey generateDataEncryptionKey() {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java

@@ -71,6 +71,7 @@
 import java.util.*;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 
 /**
@@ -1813,7 +1814,8 @@ public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
       int xmitsInProgress, int failedVolumes,
       VolumeFailureSummary volumeFailureSummary,
       @Nonnull SlowPeerReports slowPeers,
-      @Nonnull SlowDiskReports slowDisks) throws IOException {
+      @Nonnull SlowDiskReports slowDisks,
+      AtomicBoolean containsHighPriorityCmd) throws IOException {
     final DatanodeDescriptor nodeinfo;
     try {
       nodeinfo = getDatanode(nodeReg);
@@ -1922,7 +1924,8 @@ public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
     // cache commands
     addCacheCommands(blockPoolId, nodeinfo, cmds);
     // key update command
-    blockManager.addKeyUpdateCommand(cmds, nodeinfo);
+    boolean isKeyUpdated = blockManager.addKeyUpdateCommand(cmds, nodeinfo);
+    containsHighPriorityCmd.set(isKeyUpdated);
 
     // check for balancer bandwidth update
     if (nodeinfo.getBalancerBandwidth() > 0) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java

@@ -36,7 +36,7 @@
 import java.util.TreeSet;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -738,7 +738,8 @@ private void offerService() throws Exception {
             if (state == HAServiceState.ACTIVE) {
               handleRollingUpgradeStatus(resp);
             }
-            commandProcessingThread.enqueue(resp.getCommands());
+            commandProcessingThread.enqueue(resp.getCommands(),
+                resp.getIsContainsHighPriorityCmds());
             isSlownode = resp.getIsSlownode();
           }
         }
@@ -1389,7 +1390,7 @@ class CommandProcessingThread extends Thread {
     CommandProcessingThread(BPServiceActor actor) {
       super("Command processor");
       this.actor = actor;
-      this.queue = new LinkedBlockingQueue<>();
+      this.queue = new LinkedBlockingDeque<>();
       setDaemon(true);
     }
 
@@ -1468,6 +1469,11 @@ private boolean processCommand(DatanodeCommand[] cmds) {
       return true;
     }
 
+    /**
+     * Only used for cacheReport.
+     * @param cmd
+     * @throws InterruptedException
+     */
     void enqueue(DatanodeCommand cmd) throws InterruptedException {
       if (cmd == null) {
         return;
@@ -1476,6 +1482,11 @@ void enqueue(DatanodeCommand cmd) throws InterruptedException {
       dn.getMetrics().incrActorCmdQueueLength(1);
     }
 
+    /**
+     * Used for blockReport.
+     * @param cmds
+     * @throws InterruptedException
+     */
     void enqueue(List<DatanodeCommand> cmds) throws InterruptedException {
       if (cmds == null) {
         return;
@@ -1485,8 +1496,18 @@ void enqueue(List<DatanodeCommand> cmds) throws InterruptedException {
       dn.getMetrics().incrActorCmdQueueLength(1);
     }
 
-    void enqueue(DatanodeCommand[] cmds) throws InterruptedException {
-      queue.put(() -> processCommand(cmds));
+    /**
+     * Used for heartbeating.
+     * @param cmds
+     * @throws InterruptedException
+     */
+    void enqueue(DatanodeCommand[] cmds,
+        boolean containsHighPriorityCmds) throws InterruptedException {
+      if (containsHighPriorityCmds) {
+        ((LinkedBlockingDeque<Runnable>) queue).putFirst(() -> processCommand(cmds));
+      } else {
+        queue.put(() -> processCommand(cmds));
+      }
       dn.getMetrics().incrActorCmdQueueLength(1);
     }
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -97,6 +97,7 @@
 import static org.apache.hadoop.hdfs.DFSUtil.isParentEntry;
 
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.text.CaseUtils;
@@ -4429,11 +4430,12 @@ HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
           throws IOException {
     readLock();
     try {
+      AtomicBoolean containsHighPriorityCmd = new AtomicBoolean(false);
       //get datanode commands
       DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
           nodeReg, reports, getBlockPoolId(), cacheCapacity, cacheUsed,
           xceiverCount, xmitsInProgress, failedVolumes, volumeFailureSummary,
-          slowPeers, slowDisks);
+          slowPeers, slowDisks, containsHighPriorityCmd);
       long blockReportLeaseId = 0;
       if (requestFullBlockReportLease) {
         blockReportLeaseId =  blockManager.requestBlockReportLeaseId(nodeReg);
@@ -4448,7 +4450,7 @@ nodeReg, reports, getBlockPoolId(), cacheCapacity, cacheUsed,
       boolean isSlownode = slownodes.contains(nodeReg.getDatanodeUuid());
 
       return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo,
-          blockReportLeaseId, isSlownode);
+          blockReportLeaseId, isSlownode, containsHighPriorityCmd.get());
     } finally {
       readUnlock("handleHeartbeat");
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/protocol/HeartbeatResponse.java

@@ -38,21 +38,24 @@ public class HeartbeatResponse {
   private final long fullBlockReportLeaseId;
 
   private final boolean isSlownode;
+  
+  private final boolean containsHighPriorityCmds;
 
   public HeartbeatResponse(DatanodeCommand[] cmds,
       NNHAStatusHeartbeat haStatus, RollingUpgradeStatus rollingUpdateStatus,
       long fullBlockReportLeaseId) {
-    this(cmds, haStatus, rollingUpdateStatus, fullBlockReportLeaseId, false);
+    this(cmds, haStatus, rollingUpdateStatus, fullBlockReportLeaseId, false, false);
   }
 
   public HeartbeatResponse(DatanodeCommand[] cmds,
       NNHAStatusHeartbeat haStatus, RollingUpgradeStatus rollingUpdateStatus,
-      long fullBlockReportLeaseId, boolean isSlownode) {
+      long fullBlockReportLeaseId, boolean isSlownode, boolean containsHighPriorityCmds) {
     commands = cmds;
     this.haStatus = haStatus;
     this.rollingUpdateStatus = rollingUpdateStatus;
     this.fullBlockReportLeaseId = fullBlockReportLeaseId;
     this.isSlownode = isSlownode;
+    this.containsHighPriorityCmds = containsHighPriorityCmds;
   }
 
   public DatanodeCommand[] getCommands() {
@@ -74,4 +77,8 @@ public long getFullBlockReportLeaseId() {
   public boolean getIsSlownode() {
     return isSlownode;
   }
+
+  public boolean getIsContainsHighPriorityCmds() {
+    return containsHighPriorityCmds;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/proto/DatanodeProtocol.proto

@@ -224,6 +224,7 @@ message HeartbeatResponseProto {
   optional RollingUpgradeStatusProto rollingUpgradeStatusV2 = 4;
   optional uint64 fullBlockReportLeaseId = 5 [ default = 0 ];
   optional bool isSlownode = 6 [ default = false ];
+  optional bool isCommandHighPriority = 7 [ default = false ];
 }
 
 /**

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestDatanodeManager.java

@@ -35,6 +35,7 @@
 import java.util.Map.Entry;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -1039,7 +1040,7 @@ private void verifyPendingRecoveryTasks(
     Mockito.when(dm.getDatanode(dnReg)).thenReturn(nodeInfo);
     DatanodeCommand[] cmds = dm.handleHeartbeat(
         dnReg, new StorageReport[1], "bp-123", 0, 0, 10, 0, 0, null,
-        SlowPeerReports.EMPTY_REPORT, SlowDiskReports.EMPTY_REPORT);
+        SlowPeerReports.EMPTY_REPORT, SlowDiskReports.EMPTY_REPORT, new AtomicBoolean(false));
 
     long expectedNumCmds = Arrays.stream(
         new int[]{numReplicationTasks + numECTasksToBeReplicated, numECTasksToBeErasureCoded})
@@ -1171,7 +1172,7 @@ public void verifyComputeReconstructedTaskNum(int xmitsInProgress, int numReplic
 
     dm.handleHeartbeat(nodeReg, new StorageReport[1], "bp-123", 0, 0,
         10, xmitsInProgress, 0, null, SlowPeerReports.EMPTY_REPORT,
-        SlowDiskReports.EMPTY_REPORT);
+        SlowDiskReports.EMPTY_REPORT, new AtomicBoolean(false));
 
     Mockito.verify(nodeInfo).getReplicationCommand(captor.capture());
     int numReplicationTasks = captor.getValue();

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestBPOfferService.java

@@ -245,7 +245,7 @@ public HeartbeatResponse answer(InvocationOnMock invocation)
         throws Throwable {
       HeartbeatResponse heartbeatResponse = new HeartbeatResponse(
           datanodeCommands[nnIdx], mockHaStatuses[nnIdx], null,
-          0, isSlownode);
+          0, isSlownode, false);
 
       return heartbeatResponse;
     }