diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/AbstractOzoneManagerHATest.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/AbstractOzoneManagerHATest.java index 65fc84f7a54..4b25d66756b 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/AbstractOzoneManagerHATest.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/AbstractOzoneManagerHATest.java @@ -44,6 +44,7 @@ import java.util.Iterator; import java.util.UUID; import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; import org.apache.commons.io.IOUtils; import org.apache.commons.lang3.RandomStringUtils; import org.apache.hadoop.hdds.client.ReplicationFactor; @@ -86,6 +87,16 @@ public abstract class AbstractOzoneManagerHATest { private static final Duration RETRY_CACHE_DURATION = Duration.ofSeconds(30); private static OzoneClient client; + /** + * Hook for subclasses to apply extra configuration before the cluster is built. + * Call {@link #setExtraClusterConfig} in a {@code static {}} block to ensure it runs before {@code @BeforeAll}. + */ + private static Consumer extraClusterConfig = c -> { }; + + protected static void setExtraClusterConfig(Consumer config) { + extraClusterConfig = config; + } + public MiniOzoneHAClusterImpl getCluster() { return cluster; } @@ -127,6 +138,11 @@ public static Duration getRetryCacheDuration() { } protected static void initCluster(boolean followerReadEnabled) throws Exception { + initCluster(followerReadEnabled, extraClusterConfig); + } + + protected static void initCluster(boolean followerReadEnabled, + Consumer extraConfig) throws Exception { conf = new OzoneConfiguration(); omServiceId = "om-service-test1"; conf.setBoolean(OZONE_ACL_ENABLED, true); @@ -175,6 +191,8 @@ protected static void initCluster(boolean followerReadEnabled) throws Exception conf.set(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, "10s"); conf.set(OZONE_KEY_DELETING_LIMIT_PER_TASK, "2"); + extraConfig.accept(conf); + MiniOzoneHAClusterImpl.Builder clusterBuilder = MiniOzoneCluster.newHABuilder(conf) .setOMServiceId(omServiceId) .setNumOfOzoneManagers(numOfOMs); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java index e26e5328d7a..faa2d170fe3 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.om; import static java.nio.charset.StandardCharsets.UTF_8; -import static org.apache.hadoop.ozone.MiniOzoneHAClusterImpl.NODE_FAILURE_TIMEOUT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -52,6 +51,7 @@ import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadCompleteInfo; import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; import org.apache.log4j.Logger; +import org.apache.ozone.test.GenericTestUtils; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.MethodOrderer; @@ -94,7 +94,7 @@ void oneOMDown() throws Exception { changeFollowerReadInitialProxy(1); getCluster().stopOzoneManager(1); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); createVolumeTest(true); createKeyTest(true); @@ -109,7 +109,6 @@ void twoOMDown() throws Exception { getCluster().stopOzoneManager(1); getCluster().stopOzoneManager(2); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); // Write requests will fail with OMNotLeaderException createVolumeTest(false); @@ -157,7 +156,7 @@ private void testMultipartUploadWithOneOmNodeDown() throws Exception { // Stop one of the ozone manager, to see when the OM leader changes // multipart upload is happening successfully or not. getCluster().stopOzoneManager(leaderOMNodeId); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); createMultipartKeyAndReadKey(ozoneBucket, keyName, uploadID); @@ -220,11 +219,12 @@ void testLeaderOmProxyProviderFailoverOnConnectionFailure() throws Exception { // On stopping the current OM Proxy, the next connection attempt should // failover to a another OM proxy. getCluster().stopOzoneManager(firstProxyNodeId); - Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT * 4); // Next request to the proxy provider should result in a failover createVolumeTest(true); - Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT); + GenericTestUtils.waitFor( + () -> !firstProxyNodeId.equals(omFailoverProxyProvider.getCurrentProxyOMNodeId()), + 100, (int) (OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT * 5)); // Get the new OM Proxy NodeId String newProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId(); @@ -276,7 +276,6 @@ void testFollowerReadSkipsStoppedFollower() throws Exception { String stoppedFollowerNodeId = followerOmNodeIds.get(0); getCluster().stopOzoneManager(stoppedFollowerNodeId); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); followerReadFailoverProxyProvider.changeInitialProxyForTest(stoppedFollowerNodeId); objectStore.getClientProxy().listVolumes(null, null, 10); @@ -300,7 +299,7 @@ void testIncrementalWaitTimeWithSameNodeFailover() throws Exception { String leaderOMNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId(); getCluster().stopOzoneManager(leaderOMNodeId); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); createKeyTest(true); // failover should happen to new node long numTimesTriedToSameNode = omFailoverProxyProvider.getWaitTime() diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithStoppedNodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithStoppedNodes.java index 94c93d8dbe8..66c4851c3b8 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithStoppedNodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithStoppedNodes.java @@ -18,7 +18,7 @@ package org.apache.hadoop.ozone.om; import static java.nio.charset.StandardCharsets.UTF_8; -import static org.apache.hadoop.ozone.MiniOzoneHAClusterImpl.NODE_FAILURE_TIMEOUT; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -96,6 +96,10 @@ public class TestOzoneManagerHAWithStoppedNodes extends TestOzoneManagerHA { private static final org.slf4j.Logger LOG = LoggerFactory.getLogger( TestOzoneManagerHAWithStoppedNodes.class); + static { + setExtraClusterConfig(c -> c.set(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, "2s")); + } + /** * After restarting OMs we need to wait * for a leader to be elected and ready. @@ -122,7 +126,7 @@ void resetCluster() throws Exception { @Test void oneOMDown() throws Exception { getCluster().stopOzoneManager(1); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); createVolumeTest(true); createKeyTest(true); @@ -135,7 +139,6 @@ void oneOMDown() throws Exception { void twoOMDown() throws Exception { getCluster().stopOzoneManager(1); getCluster().stopOzoneManager(2); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); createVolumeTest(false); createKeyTest(false); @@ -175,7 +178,7 @@ private void testMultipartUploadWithOneOmNodeDown() throws Exception { // Stop one of the ozone manager, to see when the OM leader changes // multipart upload is happening successfully or not. getCluster().stopOzoneManager(leaderOMNodeId); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); createMultipartKeyAndReadKey(ozoneBucket, keyName, uploadID); @@ -242,11 +245,12 @@ public void testOMProxyProviderFailoverOnConnectionFailure() // On stopping the current OM Proxy, the next connection attempt should // failover to a another OM proxy. getCluster().stopOzoneManager(firstProxyNodeId); - Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT * 4); // Next request to the proxy provider should result in a failover createVolumeTest(true); - Thread.sleep(OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT); + GenericTestUtils.waitFor( + () -> !firstProxyNodeId.equals(omFailoverProxyProvider.getCurrentProxyOMNodeId()), + 100, (int) (OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT * 5)); // Get the new OM Proxy NodeId String newProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId(); @@ -354,7 +358,7 @@ void testListParts() throws Exception { // Stop leader OM, and then validate list parts. stopLeaderOM(); - Thread.sleep(NODE_FAILURE_TIMEOUT * 4); + waitForLeaderToBeReady(); validateListParts(ozoneBucket, keyName, uploadID, partsMap); @@ -438,9 +442,9 @@ public void testKeyDeletion() throws Exception { // Check on leader OM Count. GenericTestUtils.waitFor(() -> - keyDeletingService.getRunCount().get() >= 2, 10000, 120000); + keyDeletingService.getRunCount().get() >= 2, 1000, 120000); GenericTestUtils.waitFor(() -> - keyDeletingService.getDeletedKeyCount().get() == 4, 10000, 120000); + keyDeletingService.getDeletedKeyCount().get() == 4, 1000, 120000); // Check delete table is empty or not on all OMs. getCluster().getOzoneManagersList().forEach((om) -> { @@ -454,7 +458,7 @@ public void testKeyDeletion() throws Exception { return false; } }, - 10000, 120000); + 1000, 120000); } catch (Exception ex) { fail("TestOzoneManagerHAKeyDeletion failed"); } @@ -599,7 +603,7 @@ void testListVolumes() throws Exception { // Stop leader OM, and then validate list volumes for user. stopLeaderOM(); - Thread.sleep(NODE_FAILURE_TIMEOUT * 2); + waitForLeaderToBeReady(); validateVolumesList(expectedVolumes, objectStore.listVolumesByUser(userName, prefix, "")); diff --git a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java index 0a553853bd9..1e6f43b88f7 100644 --- a/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java +++ b/hadoop-ozone/mini-cluster/src/main/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java @@ -78,7 +78,7 @@ public class MiniOzoneHAClusterImpl extends MiniOzoneClusterImpl { private int waitForClusterToBeReadyTimeout = 120000; // 2 min private static final int RATIS_RPC_TIMEOUT = 1000; // 1 second - public static final int NODE_FAILURE_TIMEOUT = 2000; // 2 seconds + private static final int NODE_FAILURE_TIMEOUT = 2000; // 2 seconds public MiniOzoneHAClusterImpl( OzoneConfiguration conf,