HDDS-13614. Create separate bounded executor for ClosePipelineCommandHandler and CreatePipelineCommandHandlerHandler. (apache#8977)

aswinshakil · web-flow · commit 841f3ce752e7 · 2025-09-11T12:09:12.000+05:30
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java
@@ -26,7 +26,9 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -96,7 +98,8 @@ public class DatanodeStateMachine implements Closeable {
   private static final Logger LOG =
       LoggerFactory.getLogger(DatanodeStateMachine.class);
   private final ExecutorService executorService;
-  private final ExecutorService pipelineCommandExecutorService;
+  private final ExecutorService closePipelineCommandExecutorService;
+  private final ExecutorService createPipelineCommandExecutorService;
   private final ConfigurationSource conf;
   private final SCMConnectionManager connectionManager;
   private final ECReconstructionCoordinator ecReconstructionCoordinator;
@@ -236,11 +239,24 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
     //  datanode clients.
     DNContainerOperationClient dnClient = new DNContainerOperationClient(conf, certClient, secretKeyClient);
 
-    ThreadFactory threadFactory = new ThreadFactoryBuilder()
-        .setNameFormat(threadNamePrefix + "PipelineCommandHandlerThread-%d")
+    // Create separate bounded executors for pipeline command handlers
+    ThreadFactory closePipelineThreadFactory = new ThreadFactoryBuilder()
+        .setNameFormat(threadNamePrefix + "ClosePipelineCommandHandlerThread-%d")
         .build();
-    pipelineCommandExecutorService = Executors
-        .newSingleThreadExecutor(threadFactory);
+    closePipelineCommandExecutorService = new ThreadPoolExecutor(
+        1, 1,
+        0L, TimeUnit.MILLISECONDS,
+        new LinkedBlockingQueue<>(dnConf.getCommandQueueLimit()),
+        closePipelineThreadFactory);
+
+    ThreadFactory createPipelineThreadFactory = new ThreadFactoryBuilder()
+        .setNameFormat(threadNamePrefix + "CreatePipelineCommandHandlerThread-%d")
+        .build();
+    createPipelineCommandExecutorService = new ThreadPoolExecutor(
+        1, 1,
+        0L, TimeUnit.MILLISECONDS,
+        new LinkedBlockingQueue<>(dnConf.getCommandQueueLimit()),
+        createPipelineThreadFactory);
 
     // When we add new handlers just adding a new handler here should do the
     // trick.
@@ -257,9 +273,9 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
             dnConf.getContainerDeleteThreads(), clock,
             dnConf.getCommandQueueLimit(), threadNamePrefix))
         .addHandler(new ClosePipelineCommandHandler(conf,
-            pipelineCommandExecutorService))
+            closePipelineCommandExecutorService))
         .addHandler(new CreatePipelineCommandHandler(conf,
-            pipelineCommandExecutorService))
+            createPipelineCommandExecutorService))
         .addHandler(new SetNodeOperationalStateCommandHandler(conf,
             supervisor::nodeStateUpdated))
         .addHandler(new FinalizeNewLayoutVersionCommandHandler())
@@ -436,7 +452,8 @@ public void close() throws IOException {
     replicationSupervisorMetrics.unRegister();
     ecReconstructionMetrics.unRegister();
     executorServiceShutdownGraceful(executorService);
-    executorServiceShutdownGraceful(pipelineCommandExecutorService);
+    executorServiceShutdownGraceful(closePipelineCommandExecutorService);
+    executorServiceShutdownGraceful(createPipelineCommandExecutorService);
 
     if (connectionManager != null) {
       connectionManager.close();
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ClosePipelineCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ClosePipelineCommandHandler.java
@@ -19,8 +19,12 @@
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiFunction;
@@ -62,6 +66,7 @@ public class ClosePipelineCommandHandler implements CommandHandler {
   private final Executor executor;
   private final BiFunction<RaftPeer, GrpcTlsConfig, RaftClient> newRaftClient;
   private final MutableRate opsLatencyMs;
+  private final Set<UUID> pipelinesInProgress;
 
   /**
    * Constructs a closePipelineCommand handler.
@@ -82,6 +87,7 @@ public ClosePipelineCommandHandler(
     MetricsRegistry registry = new MetricsRegistry(
         ClosePipelineCommandHandler.class.getSimpleName());
     this.opsLatencyMs = registry.newRate(SCMCommandProto.Type.closePipelineCommand + "Ms");
+    this.pipelinesInProgress = ConcurrentHashMap.newKeySet();
   }
 
   /**
@@ -95,70 +101,88 @@ public ClosePipelineCommandHandler(
   @Override
   public void handle(SCMCommand<?> command, OzoneContainer ozoneContainer,
       StateContext context, SCMConnectionManager connectionManager) {
-    queuedCount.incrementAndGet();
-    CompletableFuture.runAsync(() -> {
-      invocationCount.incrementAndGet();
-      final long startTime = Time.monotonicNow();
-      final DatanodeDetails dn = context.getParent().getDatanodeDetails();
-      ClosePipelineCommand closePipelineCommand =
-          (ClosePipelineCommand) command;
-      final PipelineID pipelineID = closePipelineCommand.getPipelineID();
-      final HddsProtos.PipelineID pipelineIdProto = pipelineID.getProtobuf();
+    final ClosePipelineCommand closePipelineCommand = (ClosePipelineCommand) command;
+    final PipelineID pipelineID = closePipelineCommand.getPipelineID();
+    final UUID pipelineUUID = pipelineID.getId();
+    
+    // Check if this pipeline is already being processed
+    if (!pipelinesInProgress.add(pipelineUUID)) {
+      LOG.debug("Close Pipeline command for pipeline {} is already in progress, " +
+          "skipping duplicate command.", pipelineID);
+      return;
+    }
+    
+    try {
+      queuedCount.incrementAndGet();
+      CompletableFuture.runAsync(() -> {
+        invocationCount.incrementAndGet();
+        final long startTime = Time.monotonicNow();
+        final DatanodeDetails dn = context.getParent().getDatanodeDetails();
+        final HddsProtos.PipelineID pipelineIdProto = pipelineID.getProtobuf();
 
-      try {
-        XceiverServerSpi server = ozoneContainer.getWriteChannel();
-        if (server.isExist(pipelineIdProto)) {
-          if (server instanceof XceiverServerRatis) {
-            // TODO: Refactor Ratis logic to XceiverServerRatis
-            // Propagate the group remove to the other Raft peers in the pipeline
-            XceiverServerRatis ratisServer = (XceiverServerRatis) server;
-            final RaftGroupId raftGroupId = RaftGroupId.valueOf(pipelineID.getId());
-            final boolean shouldDeleteRatisLogDirectory = ratisServer.getShouldDeleteRatisLogDirectory();
-            // This might throw GroupMismatchException if the Ratis group has been closed by other datanodes
-            final Collection<RaftPeer> peers = ratisServer.getRaftPeersInPipeline(pipelineID);
-            // Try to send remove group for the other datanodes first, ignoring GroupMismatchException
-            // if the Ratis group has been closed in the other datanodes
-            peers.stream()
-                .filter(peer -> !peer.getId().equals(ratisServer.getServer().getId()))
-                .forEach(peer -> {
-                  try (RaftClient client = newRaftClient.apply(peer, ozoneContainer.getTlsClientConfig())) {
-                    client.getGroupManagementApi(peer.getId())
-                        .remove(raftGroupId, shouldDeleteRatisLogDirectory, !shouldDeleteRatisLogDirectory);
-                  } catch (GroupMismatchException ae) {
-                    // ignore silently since this means that the group has been closed by earlier close pipeline
-                    // command in another datanode
-                    LOG.debug("Failed to remove group {} for pipeline {} on peer {} since the group has " +
-                        "been removed by earlier close pipeline command handled in another datanode", raftGroupId,
-                        pipelineID, peer.getId());
-                  } catch (IOException ioe) {
-                    LOG.warn("Failed to remove group {} of pipeline {} on peer {}",
-                        raftGroupId, pipelineID, peer.getId(), ioe);
-                  }
-                });
+        try {
+          XceiverServerSpi server = ozoneContainer.getWriteChannel();
+          if (server.isExist(pipelineIdProto)) {
+            if (server instanceof XceiverServerRatis) {
+              // TODO: Refactor Ratis logic to XceiverServerRatis
+              // Propagate the group remove to the other Raft peers in the pipeline
+              XceiverServerRatis ratisServer = (XceiverServerRatis) server;
+              final RaftGroupId raftGroupId = RaftGroupId.valueOf(pipelineID.getId());
+              final boolean shouldDeleteRatisLogDirectory = ratisServer.getShouldDeleteRatisLogDirectory();
+              // This might throw GroupMismatchException if the Ratis group has been closed by other datanodes
+              final Collection<RaftPeer> peers = ratisServer.getRaftPeersInPipeline(pipelineID);
+              // Try to send remove group for the other datanodes first, ignoring GroupMismatchException
+              // if the Ratis group has been closed in the other datanodes
+              peers.stream()
+                  .filter(peer -> !peer.getId().equals(ratisServer.getServer().getId()))
+                  .forEach(peer -> {
+                    try (RaftClient client = newRaftClient.apply(peer, ozoneContainer.getTlsClientConfig())) {
+                      client.getGroupManagementApi(peer.getId())
+                          .remove(raftGroupId, shouldDeleteRatisLogDirectory, !shouldDeleteRatisLogDirectory);
+                    } catch (GroupMismatchException ae) {
+                      // ignore silently since this means that the group has been closed by earlier close pipeline
+                      // command in another datanode
+                      LOG.debug("Failed to remove group {} for pipeline {} on peer {} since the group has " +
+                          "been removed by earlier close pipeline command handled in another datanode", raftGroupId,
+                          pipelineID, peer.getId());
+                    } catch (IOException ioe) {
+                      LOG.warn("Failed to remove group {} of pipeline {} on peer {}",
+                          raftGroupId, pipelineID, peer.getId(), ioe);
+                    }
+                  });
+            }
+            // Remove the Ratis group from the current datanode pipeline, might throw GroupMismatchException as
+            // well. It is a no-op for XceiverServerSpi implementations (e.g. XceiverServerGrpc)
+            server.removeGroup(pipelineIdProto);
+            LOG.info("Close Pipeline {} command on datanode {}.", pipelineID, dn);
+          } else {
+            LOG.debug("Ignoring close pipeline command for pipeline {} on datanode {} " +
+                "as it does not exist", pipelineID, dn);
           }
-          // Remove the Ratis group from the current datanode pipeline, might throw GroupMismatchException as
-          // well. It is a no-op for XceiverServerSpi implementations (e.g. XceiverServerGrpc)
-          server.removeGroup(pipelineIdProto);
-          LOG.info("Close Pipeline {} command on datanode {}.", pipelineID, dn);
-        } else {
-          LOG.debug("Ignoring close pipeline command for pipeline {} on datanode {} " +
-              "as it does not exist", pipelineID, dn);
-        }
-      } catch (IOException e) {
-        Throwable gme = HddsClientUtils.containsException(e, GroupMismatchException.class);
-        if (gme != null) {
-          // ignore silently since this means that the group has been closed by earlier close pipeline
-          // command in another datanode
-          LOG.debug("The group for pipeline {} on datanode {} has been removed by earlier close " +
-              "pipeline command handled in another datanode", pipelineID, dn);
-        } else {
-          LOG.error("Can't close pipeline {}", pipelineID, e);
+        } catch (IOException e) {
+          Throwable gme = HddsClientUtils.containsException(e, GroupMismatchException.class);
+          if (gme != null) {
+            // ignore silently since this means that the group has been closed by earlier close pipeline
+            // command in another datanode
+            LOG.debug("The group for pipeline {} on datanode {} has been removed by earlier close " +
+                "pipeline command handled in another datanode", pipelineID, dn);
+          } else {
+            LOG.error("Can't close pipeline {}", pipelineID, e);
+          }
+        } finally {
+          long endTime = Time.monotonicNow();
+          this.opsLatencyMs.add(endTime - startTime);
         }
-      } finally {
-        long endTime = Time.monotonicNow();
-        this.opsLatencyMs.add(endTime - startTime);
-      }
-    }, executor).whenComplete((v, e) -> queuedCount.decrementAndGet());
+      }, executor).whenComplete((v, e) -> {
+        queuedCount.decrementAndGet();
+        pipelinesInProgress.remove(pipelineUUID);
+      });
+    } catch (RejectedExecutionException ex) {
+      queuedCount.decrementAndGet();
+      pipelinesInProgress.remove(pipelineUUID);
+      LOG.warn("Close Pipeline command for pipeline {} is rejected as " +
+          "command queue has reached max size.", pipelineID);
+    }
   }
 
   /**
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CreatePipelineCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/CreatePipelineCommandHandler.java
