From 766fc1820559a36e1f12932ab061f4ecf5af174f Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 9 Sep 2026 18:17:11 -0400 Subject: [PATCH 1/4] fixed transient CI failure --- .../apache/hadoop/hdds/scm/storage/TestCommitWatcher.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java index 843b738e6e90..10ffdbff1f0e 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java @@ -128,6 +128,11 @@ public void init() throws Exception { .setNumDatanodes(5) .build(); cluster.waitForClusterToBeReady(); + // Wait for the RATIS THREE pipeline to reach OPEN state before any writes. + // A pipeline only opens once it is healthy, which requires an elected Ratis + // leader; otherwise the first write can race leader election and fail with + // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. + cluster.waitForPipelineTobeReady(HddsProtos.ReplicationFactor.THREE, 60000); client = OzoneClientFactory.getRpcClient(conf); ObjectStore objectStore = client.getObjectStore(); objectStore.createVolume(VOLUME_NAME); From a557792d74d2032590c20cf577323b79a1605c35 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Mon, 14 Sep 2026 14:26:47 -0400 Subject: [PATCH 2/4] addressed review comments --- .../hdds/scm/storage/TestCommitWatcher.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java index 10ffdbff1f0e..b6c0e66681a4 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java @@ -61,6 +61,7 @@ import org.apache.hadoop.ozone.client.OzoneClientFactory; import org.apache.hadoop.ozone.common.ChunkBuffer; import org.apache.hadoop.ozone.container.ContainerTestHelper; +import org.apache.hadoop.ozone.container.OzoneTestHelper; import org.apache.ratis.protocol.exceptions.AlreadyClosedException; import org.apache.ratis.protocol.exceptions.NotReplicatedException; import org.apache.ratis.protocol.exceptions.RaftRetryFailureException; @@ -162,6 +163,15 @@ public void testReleaseBuffers() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); + // Ensure the freshly-allocated pipeline has an elected Ratis leader + // before the async write burst: register the Ratis group on the + // pipeline datanodes and commit a CreateContainer synchronously. + // Otherwise the first write races leader election and can fail with + // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. + OzoneTestHelper.createPipelineOnDatanode(pipeline, cluster); + ratisClient.sendCommandAsync( + ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) + .getResponse().get(); CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient); BlockID blockID = ContainerTestHelper.getTestBlockID(containerId); List replies = new ArrayList<>(); @@ -228,6 +238,15 @@ public void testReleaseBuffersOnException() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); + // Ensure the freshly-allocated pipeline has an elected Ratis leader + // before the async write burst: register the Ratis group on the + // pipeline datanodes and commit a CreateContainer synchronously. + // Otherwise the first write races leader election and can fail with + // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. + OzoneTestHelper.createPipelineOnDatanode(pipeline, cluster); + ratisClient.sendCommandAsync( + ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) + .getResponse().get(); CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient); BlockID blockID = ContainerTestHelper.getTestBlockID(containerId); List replies = new ArrayList<>(); From 3465e0c76323c748ec9428195d5ca623596d7f96 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 16 Sep 2026 11:22:08 -0400 Subject: [PATCH 3/4] addressed review comments --- .../hdds/scm/storage/TestCommitWatcher.java | 40 ++++++++++--------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java index b6c0e66681a4..8d5b29017fee 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java @@ -61,7 +61,6 @@ import org.apache.hadoop.ozone.client.OzoneClientFactory; import org.apache.hadoop.ozone.common.ChunkBuffer; import org.apache.hadoop.ozone.container.ContainerTestHelper; -import org.apache.hadoop.ozone.container.OzoneTestHelper; import org.apache.ratis.protocol.exceptions.AlreadyClosedException; import org.apache.ratis.protocol.exceptions.NotReplicatedException; import org.apache.ratis.protocol.exceptions.RaftRetryFailureException; @@ -129,10 +128,9 @@ public void init() throws Exception { .setNumDatanodes(5) .build(); cluster.waitForClusterToBeReady(); - // Wait for the RATIS THREE pipeline to reach OPEN state before any writes. - // A pipeline only opens once it is healthy, which requires an elected Ratis - // leader; otherwise the first write can race leader election and fail with - // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. + // Make sure background pipeline creation has finished and a RATIS THREE + // pipeline is OPEN before allocating a container, so the write burst does + // not race on-demand pipeline creation inside allocateContainer. cluster.waitForPipelineTobeReady(HddsProtos.ReplicationFactor.THREE, 60000); client = OzoneClientFactory.getRpcClient(conf); ObjectStore objectStore = client.getObjectStore(); @@ -163,22 +161,24 @@ public void testReleaseBuffers() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); - // Ensure the freshly-allocated pipeline has an elected Ratis leader - // before the async write burst: register the Ratis group on the - // pipeline datanodes and commit a CreateContainer synchronously. - // Otherwise the first write races leader election and can fail with - // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. - OzoneTestHelper.createPipelineOnDatanode(pipeline, cluster); + // Warm up the freshly-acquired Ratis client before the async write + // burst: create the container synchronously so the RaftClient discovers + // the leader and the container already exists. Otherwise the first async + // WriteChunk has to do leader discovery and lazy container creation under + // the aggressive request/no-retry timeouts and can flake with + // AlreadyClosedException. ratisClient.sendCommandAsync( ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) .getResponse().get(); CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient); - BlockID blockID = ContainerTestHelper.getTestBlockID(containerId); List replies = new ArrayList<>(); long length = 0; List> 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); @@ -238,22 +238,24 @@ public void testReleaseBuffersOnException() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); - // Ensure the freshly-allocated pipeline has an elected Ratis leader - // before the async write burst: register the Ratis group on the - // pipeline datanodes and commit a CreateContainer synchronously. - // Otherwise the first write races leader election and can fail with - // NotLeaderException -> RaftRetryFailureException -> AlreadyClosedException. - OzoneTestHelper.createPipelineOnDatanode(pipeline, cluster); + // Warm up the freshly-acquired Ratis client before the async write + // burst: create the container synchronously so the RaftClient discovers + // the leader and the container already exists. Otherwise the first async + // WriteChunk has to do leader discovery and lazy container creation under + // the aggressive request/no-retry timeouts and can flake with + // AlreadyClosedException. ratisClient.sendCommandAsync( ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) .getResponse().get(); CommitWatcher watcher = new CommitWatcher(bufferPool, ratisClient); - BlockID blockID = ContainerTestHelper.getTestBlockID(containerId); List replies = new ArrayList<>(); long length = 0; List> 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); From 010c5eaabfd92ada179dac79d8f42f364cf23749 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 16 Sep 2026 11:46:22 -0400 Subject: [PATCH 4/4] removed unneeded changes --- .../hdds/scm/storage/TestCommitWatcher.java | 24 +++++++------------ 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java index 8d5b29017fee..0eddf562bed1 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/storage/TestCommitWatcher.java @@ -128,10 +128,6 @@ public void init() throws Exception { .setNumDatanodes(5) .build(); cluster.waitForClusterToBeReady(); - // Make sure background pipeline creation has finished and a RATIS THREE - // pipeline is OPEN before allocating a container, so the write burst does - // not race on-demand pipeline creation inside allocateContainer. - cluster.waitForPipelineTobeReady(HddsProtos.ReplicationFactor.THREE, 60000); client = OzoneClientFactory.getRpcClient(conf); ObjectStore objectStore = client.getObjectStore(); objectStore.createVolume(VOLUME_NAME); @@ -161,12 +157,10 @@ public void testReleaseBuffers() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); - // Warm up the freshly-acquired Ratis client before the async write - // burst: create the container synchronously so the RaftClient discovers - // the leader and the container already exists. Otherwise the first async - // WriteChunk has to do leader discovery and lazy container creation under - // the aggressive request/no-retry timeouts and can flake with - // AlreadyClosedException. + // Commit one request synchronously before the async write burst so the + // Ratis client is fully established and the container already exists. + // Issuing the burst directly on a freshly-acquired client was seen to + // fail intermittently with AlreadyClosedException (HDDS-16398). ratisClient.sendCommandAsync( ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) .getResponse().get(); @@ -238,12 +232,10 @@ public void testReleaseBuffersOnException() throws Exception { try (XceiverClientSpi xceiverClient = mgr.acquireClient(pipeline)) { assertEquals(1, xceiverClient.getRefcount()); XceiverClientRatis ratisClient = assertInstanceOf(XceiverClientRatis.class, xceiverClient); - // Warm up the freshly-acquired Ratis client before the async write - // burst: create the container synchronously so the RaftClient discovers - // the leader and the container already exists. Otherwise the first async - // WriteChunk has to do leader discovery and lazy container creation under - // the aggressive request/no-retry timeouts and can flake with - // AlreadyClosedException. + // Commit one request synchronously before the async write burst so the + // Ratis client is fully established and the container already exists. + // Issuing the burst directly on a freshly-acquired client was seen to + // fail intermittently with AlreadyClosedException (HDDS-16398). ratisClient.sendCommandAsync( ContainerTestHelper.getCreateContainerRequest(containerId, pipeline)) .getResponse().get();