HDDS-420. putKey failing with KEY_ALLOCATION_ERROR. Contributed by Shashikant Banerjee.

(cherry picked from commit c6e19db197253b660c8b16854360fe95b1d7c43f)

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

@@ -280,7 +280,7 @@ public class StateContext {
             && pipelineActionIter.hasClosePipeline() && pipelineAction
             .hasClosePipeline()
             && pipelineActionIter.getClosePipeline().getPipelineID()
-            == pipelineAction.getClosePipeline().getPipelineID()) {
+            .equals(pipelineAction.getClosePipeline().getPipelineID())) {
           return;
         }
       }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java

@@ -397,6 +397,9 @@ public final class XceiverServerRatis implements XceiverServerSpi {
         .setAction(PipelineAction.Action.CLOSE)
         .build();
     context.addPipelineActionIfAbsent(action);
+    LOG.debug(
+        "pipeline Action " + action.getAction() + "  on pipeline " + pipelineID
+            + ".Reason : " + action.getClosePipeline().getDetailedReason());
   }
 
   void handleNodeSlowness(

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/StaleNodeHandler.java

@@ -25,6 +25,8 @@ import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.node.states.Node2ContainerMap;
 import org.apache.hadoop.hdds.server.events.EventHandler;
 import org.apache.hadoop.hdds.server.events.EventPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Set;
 
@@ -32,6 +34,7 @@ import java.util.Set;
  * Handles Stale node event.
  */
 public class StaleNodeHandler implements EventHandler<DatanodeDetails> {
+  static final Logger LOG = LoggerFactory.getLogger(StaleNodeHandler.class);
 
   private final Node2ContainerMap node2ContainerMap;
   private final Mapping containerManager;
@@ -48,6 +51,7 @@ public class StaleNodeHandler implements EventHandler<DatanodeDetails> {
     Set<PipelineID> pipelineIDs =
         containerManager.getPipelineOnDatanode(datanodeDetails);
     for (PipelineID id : pipelineIDs) {
+      LOG.info("closing pipeline {}.", id);
       publisher.fireEvent(SCMEvents.PIPELINE_CLOSE, id);
     }
   }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipelines/PipelineActionEventHandler.java

@@ -50,6 +50,8 @@ public class PipelineActionEventHandler implements
       case CLOSE:
         PipelineID pipelineID = PipelineID.
             getFromProtobuf(action.getClosePipeline().getPipelineID());
+        LOG.info("Closing pipeline " + pipelineID + " for reason:" + action
+            .getClosePipeline().getDetailedReason());
         publisher.fireEvent(SCMEvents.PIPELINE_CLOSE, pipelineID);
         break;
       default:

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipelines/ratis/RatisManagerImpl.java

@@ -110,15 +110,15 @@ public class RatisManagerImpl extends PipelineManager {
    * Close the pipeline.
    */
   public void closePipeline(Pipeline pipeline) throws IOException {
+    try (XceiverClientRatis client =
+        XceiverClientRatis.newXceiverClientRatis(pipeline, conf)) {
+      client.destroyPipeline();
+    }
     super.closePipeline(pipeline);
     for (DatanodeDetails node : pipeline.getMachines()) {
       // A node should always be the in ratis members list.
       Preconditions.checkArgument(ratisMembers.remove(node));
     }
-    try (XceiverClientRatis client =
-        XceiverClientRatis.newXceiverClientRatis(pipeline, conf)) {
-      client.destroyPipeline();
-    }
   }
 
   /**