HDDS-16398. Fix intermittent AlreadyClosedException in TestCommitWatcher

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.DatanodeRatisServerConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -56,11 +57,13 @@
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.RatisTestHelper;
 import org.apache.hadoop.ozone.client.ObjectStore;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneClientFactory;
 import org.apache.hadoop.ozone.common.ChunkBuffer;
 import org.apache.hadoop.ozone.container.ContainerTestHelper;
+import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
 import org.apache.ratis.protocol.exceptions.NotReplicatedException;
 import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
@@ -144,26 +147,59 @@ public void shutdown() {
     }
   }
 
+  /**
+   * Allocates a RATIS THREE container and waits until SCM's record of its pipeline names the datanode that
+   * currently leads the Raft group.
+   * <p>
+   * SCM learns a pipeline's leader from datanode pipeline reports, which ride the heartbeat (10s here), and
+   * never unsets it, so {@link Pipeline#getLeaderNode()} can name a datanode that has since stopped leading.
+   * The Ratis client aims its first request at that node. Ratis puts a dummy watch request at the head of the
+   * ordered sliding window, so a first request that lands on a follower burns the watch retry budget - only
+   * 10s in this test - and fails every request queued behind it with AlreadyClosedException (HDDS-16398).
+   */
+  private ContainerWithPipeline allocateContainerWithElectedLeader() throws Exception {
+    ContainerWithPipeline allocated = storageContainerLocationClient.allocateContainer(
+        HddsProtos.ReplicationType.RATIS, HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE);
+    long containerId = allocated.getContainerInfo().getContainerID();
+    AtomicReference<ContainerWithPipeline> withElectedLeader = new AtomicReference<>();
+    GenericTestUtils.waitFor(() -> {
+      try {
+        ContainerWithPipeline current = storageContainerLocationClient.getContainerWithPipeline(containerId);
+        Pipeline pipeline = current.getPipeline();
+        if (pipeline.getLeaderId() == null
+            || !RatisTestHelper.isRatisLeader(cluster.getHddsDatanode(pipeline.getLeaderNode()), pipeline)) {
+          return false;
+        }
+        withElectedLeader.set(current);
+        return true;
+      } catch (Exception e) {
+        // SCM or the datanode is not ready to answer yet; keep polling.
+        return false;
+      }
+    }, 500, 60000);
+    return withElectedLeader.get();
+  }
+
   @Test
   public void testReleaseBuffers() throws Exception {
     int capacity = 2;
     BufferPool bufferPool = new BufferPool(CHUNK_SIZE, capacity);
     try (XceiverClientManager mgr = new XceiverClientManager(conf)) {
-      ContainerWithPipeline container = storageContainerLocationClient
-          .allocateContainer(HddsProtos.ReplicationType.RATIS,
-              HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE);
+      ContainerWithPipeline container = allocateContainerWithElectedLeader();
       Pipeline pipeline = container.getPipeline();
       long containerId = container.getContainerInfo().getContainerID();
       try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) {
         assertEquals(1, xceiverClient.getRefcount());
         XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient);
         CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient);
-        BlockID blockID = ContainerTestHelper.getTestBlockID(containerId);
         List<XceiverClientReply> replies = new ArrayList<>();
         long length = 0;
         List<CompletableFuture<ContainerCommandResponseProto>>
             futures = new ArrayList<>();
         for (int i = 0; i < capacity; i++) {
+          // Use a distinct block per iteration; a real client never rewrites the
+          // same blockID and offset with different data.
+          BlockID blockID = ContainerTestHelper.getTestBlockID(containerId);
           ContainerCommandRequestProto writeChunkRequest =
               ContainerTestHelper
                   .getWriteChunkRequest(pipeline, blockID, CHUNK_SIZE);
@@ -215,21 +251,21 @@ public void testReleaseBuffersOnException() throws Exception {
     int capacity = 2;
     BufferPool bufferPool = new BufferPool(CHUNK_SIZE, capacity);
     try (XceiverClientManager mgr = new XceiverClientManager(conf)) {
-      ContainerWithPipeline container = storageContainerLocationClient
-          .allocateContainer(HddsProtos.ReplicationType.RATIS,
-              HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE);
+      ContainerWithPipeline container = allocateContainerWithElectedLeader();
       Pipeline pipeline = container.getPipeline();
       long containerId = container.getContainerInfo().getContainerID();
       try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) {
         assertEquals(1, xceiverClient.getRefcount());
         XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient);
         CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient);
-        BlockID blockID = ContainerTestHelper.getTestBlockID(containerId);
         List<XceiverClientReply> replies = new ArrayList<>();
         long length = 0;
         List<CompletableFuture<ContainerCommandResponseProto>>
             futures = new ArrayList<>();
         for (int i = 0; i < capacity; i++) {
+          // Use a distinct block per iteration; a real client never rewrites the
+          // same blockID and offset with different data.
+          BlockID blockID = ContainerTestHelper.getTestBlockID(containerId);
           ContainerCommandRequestProto writeChunkRequest =
               ContainerTestHelper
                   .getWriteChunkRequest(pipeline, blockID, CHUNK_SIZE);