You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uniffle.apache.org by ya...@apache.org on 2025/03/12 13:36:44 UTC
(uniffle) branch master updated: [#2385] improvement(test): Make all tests in `integration-common-test` use random port (#2387)
This is an automated email from the ASF dual-hosted git repository.
yangjie01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/uniffle.git
The following commit(s) were added to refs/heads/master by this push:
new 87362fb70 [#2385] improvement(test): Make all tests in `integration-common-test` use random port (#2387)
87362fb70 is described below
commit 87362fb70970c7d5757863cadb8ce23eb3d59eb8
Author: summaryzb <su...@gmail.com>
AuthorDate: Wed Mar 12 21:36:39 2025 +0800
[#2385] improvement(test): Make all tests in `integration-common-test` use random port (#2387)
### What changes were proposed in this pull request?
All tests in integration-test/common use random port
### Why are the changes needed?
Fix: #2385
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
UT
---
.../uniffle/test/AssignmentWithTagsTest.java | 105 +++++++-------------
.../uniffle/test/CoordinatorAdminServiceTest.java | 12 +--
.../uniffle/test/CoordinatorAssignmentTest.java | 33 +++----
.../uniffle/test/CoordinatorGrpcServerTest.java | 10 +-
.../apache/uniffle/test/CoordinatorGrpcTest.java | 60 ++++++------
.../test/CoordinatorReconfigureNodeMaxTest.java | 29 ++----
.../apache/uniffle/test/CoordinatorTestBase.java | 12 ++-
.../uniffle/test/DiskErrorToleranceTest.java | 40 +++-----
.../apache/uniffle/test/FetchClientConfTest.java | 52 +++++-----
.../test/HealthCheckCoordinatorGrpcTest.java | 48 +++++-----
.../org/apache/uniffle/test/HealthCheckTest.java | 2 +-
.../test/HybridStorageFaultToleranceBase.java | 17 ++--
.../test/HybridStorageHadoopFallbackTest.java | 45 ++-------
.../test/HybridStorageLocalFileFallbackTest.java | 35 +++----
.../apache/uniffle/test/IntegrationTestBase.java | 61 ++++++++----
.../test/PartitionBalanceCoordinatorGrpcTest.java | 51 +++++-----
.../java/org/apache/uniffle/test/QuorumTest.java | 106 ++++++---------------
.../test/RemoteMergeShuffleWithRssClientTest.java | 47 +++------
...ShuffleWithRssClientTestWhenShuffleFlushed.java | 48 +++-------
.../apache/uniffle/test/RpcClientRetryTest.java | 51 +++-------
.../apache/uniffle/test/ShuffleReadWriteBase.java | 7 --
.../ShuffleServerConcurrentWriteOfHadoopTest.java | 33 +++----
.../test/ShuffleServerFaultToleranceTest.java | 3 +-
.../apache/uniffle/test/ShuffleServerGrpcTest.java | 49 ++++------
.../test/ShuffleServerInternalGrpcTest.java | 25 +++--
.../test/ShuffleServerOnRandomPortTest.java | 48 +++++-----
.../uniffle/test/ShuffleServerWithHadoopTest.java | 46 ++++-----
.../ShuffleServerWithKerberizedHadoopTest.java | 14 +--
.../ShuffleServerWithLocalOfExceptionTest.java | 17 +---
.../ShuffleServerWithLocalOfLocalOrderTest.java | 44 +++------
.../uniffle/test/ShuffleServerWithLocalTest.java | 43 +++------
.../test/ShuffleServerWithMemLocalHadoopTest.java | 46 +++------
.../uniffle/test/ShuffleServerWithMemoryTest.java | 50 ++++------
.../uniffle/test/ShuffleWithRssClientTest.java | 61 +++++-------
34 files changed, 506 insertions(+), 844 deletions(-)
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java
index 2512c6c05..6b84a28b6 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java
@@ -18,9 +18,7 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -32,8 +30,6 @@ import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.uniffle.client.factory.ShuffleClientFactory;
import org.apache.uniffle.client.impl.ShuffleWriteClientImpl;
@@ -42,7 +38,6 @@ import org.apache.uniffle.common.ShuffleAssignmentsInfo;
import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.common.util.Constants;
import org.apache.uniffle.coordinator.CoordinatorConf;
-import org.apache.uniffle.coordinator.CoordinatorServer;
import org.apache.uniffle.server.ShuffleServer;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
@@ -56,87 +51,51 @@ import static org.junit.jupiter.api.Assertions.fail;
* {@code RssClientConfig.RSS_CLIENT_ASSIGNMENT_TAGS}
*/
public class AssignmentWithTagsTest extends CoordinatorTestBase {
- private static final Logger LOG = LoggerFactory.getLogger(AssignmentWithTagsTest.class);
// KV: tag -> shuffle server id
private static Map<String, List<Integer>> tagOfShufflePorts = new HashMap<>();
+ private static final String tag1 = "fixed";
+ private static final String tag2 = "elastic";
- private static List<Integer> findAvailablePorts(int num) throws IOException {
- List<ServerSocket> sockets = new ArrayList<>();
- List<Integer> ports = new ArrayList<>();
-
- for (int i = 0; i < num; i++) {
- ServerSocket socket = new ServerSocket(0);
- ports.add(socket.getLocalPort());
- sockets.add(socket);
- }
-
- for (ServerSocket socket : sockets) {
- socket.close();
- }
-
- return ports;
- }
-
- private static void createAndStartShuffleServerWithTags(Set<String> tags, File tmpDir)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
+ private static void prepareShuffleServerConf(int subDirIndex, Set<String> tags, File tmpDir) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tmpDir, ServerType.GRPC);
shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 4000);
-
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
-
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
shuffleServerConf.setString("rss.server.tags", StringUtils.join(tags, ","));
-
- List<Integer> ports = findAvailablePorts(2);
- shuffleServerConf.setInteger("rss.rpc.server.port", ports.get(0));
- shuffleServerConf.setInteger("rss.jetty.http.port", ports.get(1));
-
- for (String tag : tags) {
- tagOfShufflePorts.putIfAbsent(tag, new ArrayList<>());
- tagOfShufflePorts.get(tag).add(ports.get(0));
- }
- tagOfShufflePorts.putIfAbsent(Constants.SHUFFLE_SERVER_VERSION, new ArrayList<>());
- tagOfShufflePorts.get(Constants.SHUFFLE_SERVER_VERSION).add(ports.get(0));
-
- LOG.info(
- "Shuffle server data dir: {}, rpc port: {}, http port: {}",
- dataDir1 + "," + dataDir2,
- ports.get(0),
- ports.get(1));
-
- ShuffleServer server = new ShuffleServer(shuffleServerConf);
- grpcShuffleServers.add(server);
- server.start();
+ storeShuffleServerConf(shuffleServerConf);
}
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- for (CoordinatorServer coordinator : coordinators) {
- coordinator.start();
- }
-
- File dir1 = new File(tmpDir, "server1");
for (int i = 0; i < 2; i++) {
- createAndStartShuffleServerWithTags(Sets.newHashSet(), dir1);
+ prepareShuffleServerConf(i, Sets.newHashSet(), tmpDir);
}
- File dir2 = new File(tmpDir, "server2");
for (int i = 0; i < 2; i++) {
- createAndStartShuffleServerWithTags(Sets.newHashSet("fixed"), dir2);
+ prepareShuffleServerConf(2 + i, Sets.newHashSet(tag1), tmpDir);
}
- File dir3 = new File(tmpDir, "server3");
for (int i = 0; i < 2; i++) {
- createAndStartShuffleServerWithTags(Sets.newHashSet("elastic"), dir3);
+ prepareShuffleServerConf(4 + i, Sets.newHashSet(tag2), tmpDir);
}
+ startServersWithRandomPorts();
+ List<Integer> collect =
+ grpcShuffleServers.stream().map(ShuffleServer::getGrpcPort).collect(Collectors.toList());
+ tagOfShufflePorts.put(Constants.SHUFFLE_SERVER_VERSION, collect);
+ tagOfShufflePorts.put(
+ tag1,
+ Arrays.asList(
+ grpcShuffleServers.get(2).getGrpcPort(), grpcShuffleServers.get(3).getGrpcPort()));
+ tagOfShufflePorts.put(
+ tag2,
+ Arrays.asList(
+ grpcShuffleServers.get(4).getGrpcPort(), grpcShuffleServers.get(5).getGrpcPort()));
+
// Wait all shuffle servers registering to coordinator
long startTimeMS = System.currentTimeMillis();
while (true) {
@@ -151,7 +110,7 @@ public class AssignmentWithTagsTest extends CoordinatorTestBase {
}
@Test
- public void testTags() throws Exception {
+ public void testTags() {
ShuffleWriteClientImpl shuffleWriteClient =
ShuffleClientFactory.newWriteBuilder()
.clientType(ClientType.GRPC.name())
@@ -168,7 +127,7 @@ public class AssignmentWithTagsTest extends CoordinatorTestBase {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
- shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
+ shuffleWriteClient.registerCoordinators(getQuorum());
// Case1 : only set the single default shuffle version tag
ShuffleAssignmentsInfo assignmentsInfo =
@@ -198,26 +157,26 @@ public class AssignmentWithTagsTest extends CoordinatorTestBase {
// Case3: Set the single fixed tag
assignmentsInfo =
- shuffleWriteClient.getShuffleAssignments("app-3", 1, 1, 1, Sets.newHashSet("fixed"), 1, -1);
+ shuffleWriteClient.getShuffleAssignments("app-3", 1, 1, 1, Sets.newHashSet(tag1), 1, -1);
assignedServerPorts =
assignmentsInfo.getPartitionToServers().values().stream()
.flatMap(x -> x.stream())
.map(x -> x.getGrpcPort())
.collect(Collectors.toList());
assertEquals(1, assignedServerPorts.size());
- assertTrue(tagOfShufflePorts.get("fixed").contains(assignedServerPorts.get(0)));
+ assertTrue(tagOfShufflePorts.get(tag1).contains(assignedServerPorts.get(0)));
// case4: Set the multiple tags if exists
assignmentsInfo =
shuffleWriteClient.getShuffleAssignments(
- "app-4", 1, 1, 1, Sets.newHashSet("fixed", Constants.SHUFFLE_SERVER_VERSION), 1, -1);
+ "app-4", 1, 1, 1, Sets.newHashSet(tag1, Constants.SHUFFLE_SERVER_VERSION), 1, -1);
assignedServerPorts =
assignmentsInfo.getPartitionToServers().values().stream()
.flatMap(x -> x.stream())
.map(x -> x.getGrpcPort())
.collect(Collectors.toList());
assertEquals(1, assignedServerPorts.size());
- assertTrue(tagOfShufflePorts.get("fixed").contains(assignedServerPorts.get(0)));
+ assertTrue(tagOfShufflePorts.get(tag1).contains(assignedServerPorts.get(0)));
// case5: Set the multiple tags if non-exist
try {
@@ -227,7 +186,7 @@ public class AssignmentWithTagsTest extends CoordinatorTestBase {
1,
1,
1,
- Sets.newHashSet("fixed", "elastic", Constants.SHUFFLE_SERVER_VERSION),
+ Sets.newHashSet(tag1, tag2, Constants.SHUFFLE_SERVER_VERSION),
1,
-1);
fail();
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java
index 0b8e21a03..17e87e0b0 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java
@@ -34,8 +34,6 @@ import org.apache.uniffle.coordinator.CoordinatorConf;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class CoordinatorAdminServiceTest extends IntegrationTestBase {
-
- private static final Integer JETTY_HTTP_PORT = 12345;
private static final String accessChecker =
"org.apache.uniffle.test.AccessClusterTest$MockedAccessChecker";
@@ -45,18 +43,16 @@ public class CoordinatorAdminServiceTest extends IntegrationTestBase {
@BeforeAll
public static void setUp() throws Exception {
- CoordinatorConf coordinatorConf = new CoordinatorConf();
- coordinatorConf.set(RssBaseConf.JETTY_HTTP_PORT, JETTY_HTTP_PORT);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.set(RssBaseConf.JETTY_CORE_POOL_SIZE, 128);
- coordinatorConf.set(RssBaseConf.RPC_SERVER_PORT, 12346);
coordinatorConf.setString(CoordinatorConf.COORDINATOR_ACCESS_CHECKERS.key(), accessChecker);
- createCoordinatorServer(coordinatorConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+ startServersWithRandomPorts();
}
@BeforeEach
public void createClient() {
- String hostUrl = String.format("http://%s:%d", LOCALHOST, JETTY_HTTP_PORT);
+ String hostUrl = String.format("http://%s:%d", LOCALHOST, jettyPorts.get(0));
adminRestApi = new AdminRestApi(UniffleRestClient.builder(hostUrl).build());
}
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java
index fd45c6929..136cc22e8 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java
@@ -19,7 +19,6 @@ package org.apache.uniffle.test;
import java.io.File;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -54,42 +53,32 @@ public class CoordinatorAssignmentTest extends CoordinatorTestBase {
private static final int SERVER_NUM = 10;
private static final HashSet<String> TAGS = Sets.newHashSet("t1");
- private static final String QUORUM =
- LOCALHOST + ":" + COORDINATOR_PORT_1 + "," + LOCALHOST + ":" + COORDINATOR_PORT_2;
-
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf1 = getCoordinatorConf();
+ CoordinatorConf coordinatorConf1 = coordinatorConfWithoutPort();
coordinatorConf1.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
coordinatorConf1.setInteger(CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, SHUFFLE_NODES_MAX);
coordinatorConf1.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
- createCoordinatorServer(coordinatorConf1);
+ storeCoordinatorConf(coordinatorConf1);
- CoordinatorConf coordinatorConf2 = getCoordinatorConf();
+ CoordinatorConf coordinatorConf2 = coordinatorConfWithoutPort();
coordinatorConf2.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
coordinatorConf2.setInteger(CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, SHUFFLE_NODES_MAX);
- coordinatorConf2.setInteger(CoordinatorConf.RPC_SERVER_PORT, COORDINATOR_PORT_2);
- coordinatorConf2.setInteger(CoordinatorConf.JETTY_HTTP_PORT, JETTY_PORT_2);
coordinatorConf2.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
- createCoordinatorServer(coordinatorConf2);
+ storeCoordinatorConf(coordinatorConf2);
for (int i = 0; i < SERVER_NUM; i++) {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- File dataDir = new File(tmpDir, "data" + i);
- String basePath = dataDir.getAbsolutePath();
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(i, tmpDir, ServerType.GRPC);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.MEMORY_LOCALFILE_HDFS.name());
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 2000L);
shuffleServerConf.set(ShuffleServerConf.SERVER_PRE_ALLOCATION_EXPIRED, 5000L);
- shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 18001 + i);
- shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, 19010 + i);
shuffleServerConf.set(ShuffleServerConf.TAGS, new ArrayList<>(TAGS));
- shuffleServerConf.setString("rss.coordinator.quorum", QUORUM);
- createShuffleServer(shuffleServerConf);
+ storeShuffleServerConf(shuffleServerConf);
}
- startServers();
+ startServersWithRandomPorts();
Thread.sleep(1000 * 5);
}
@@ -112,7 +101,7 @@ public class CoordinatorAssignmentTest extends CoordinatorTestBase {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
- shuffleWriteClient.registerCoordinators(QUORUM);
+ shuffleWriteClient.registerCoordinators(getQuorum());
// Case1: Disable silent period
ShuffleAssignmentsInfo info =
@@ -154,7 +143,7 @@ public class CoordinatorAssignmentTest extends CoordinatorTestBase {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
- shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
+ shuffleWriteClient.registerCoordinators(getQuorum());
/**
* case1: user specify the illegal shuffle node num(<0) it will use the default shuffle nodes
@@ -204,7 +193,7 @@ public class CoordinatorAssignmentTest extends CoordinatorTestBase {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
- shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
+ shuffleWriteClient.registerCoordinators(getQuorum());
Set<String> excludeServer = Sets.newConcurrentHashSet();
List<ShuffleServer> excludeShuffleServer =
grpcShuffleServers.stream().limit(3).collect(Collectors.toList());
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java
index cc75624b8..384b07f38 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java
@@ -58,7 +58,7 @@ public class CoordinatorGrpcServerTest {
@Test
public void testGrpcConnectionSize() throws Exception {
RssBaseConf baseConf = new RssBaseConf();
- baseConf.set(RssBaseConf.RPC_SERVER_PORT, 20001);
+ baseConf.set(RssBaseConf.RPC_SERVER_PORT, 0);
baseConf.set(RssBaseConf.RPC_EXECUTOR_SIZE, 2);
GRPCMetrics grpcMetrics = new CoordinatorGrpcMetrics(baseConf);
@@ -70,7 +70,7 @@ public class CoordinatorGrpcServerTest {
.addService(new MockedCoordinatorGrpcService())
.build();
try {
- grpcServer.start();
+ final int port = grpcServer.start();
// case1: test the single one connection metric
double connSize = grpcMetrics.getGaugeMap().get(GRPC_SERVER_CONNECTION_NUMBER_KEY).get();
@@ -78,9 +78,9 @@ public class CoordinatorGrpcServerTest {
// case2: test the multiple connections
try (CoordinatorGrpcClient coordinatorGrpcClient =
- new CoordinatorGrpcClient("localhost", 20001);
- CoordinatorGrpcClient client1 = new CoordinatorGrpcClient("localhost", 20001);
- CoordinatorGrpcClient client2 = new CoordinatorGrpcClient("localhost", 20001)) {
+ new CoordinatorGrpcClient("localhost", port);
+ CoordinatorGrpcClient client1 = new CoordinatorGrpcClient("localhost", port);
+ CoordinatorGrpcClient client2 = new CoordinatorGrpcClient("localhost", port)) {
coordinatorGrpcClient.registerApplicationInfo(
new RssApplicationInfoRequest("testGrpcConnectionSize", 10000, "user"));
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcTest.java
index cd63f4787..cacc46c54 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcTest.java
@@ -17,17 +17,17 @@
package org.apache.uniffle.test;
-import java.lang.reflect.Field;
+import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.apache.uniffle.client.request.RssApplicationInfoRequest;
import org.apache.uniffle.client.request.RssGetShuffleAssignmentsRequest;
@@ -38,7 +38,6 @@ import org.apache.uniffle.common.PartitionRange;
import org.apache.uniffle.common.ShuffleRegisterInfo;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.config.RssBaseConf;
-import org.apache.uniffle.common.config.RssConf;
import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.common.rpc.StatusCode;
import org.apache.uniffle.common.storage.StorageInfo;
@@ -67,20 +66,18 @@ import static uk.org.webcompere.systemstubs.SystemStubs.withEnvironmentVariables
public class CoordinatorGrpcTest extends CoordinatorTestBase {
@BeforeAll
- public static void setupServers() throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ public static void setupServers(@TempDir File tmpDir) throws Exception {
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
coordinatorConf.setString(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY.key(), "BASIC");
coordinatorConf.setLong("rss.coordinator.app.expired", 2000);
coordinatorConf.setLong("rss.coordinator.server.heartbeat.timeout", 3000);
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.remove(ShuffleServerConf.NETTY_SERVER_PORT.key());
- createShuffleServer(shuffleServerConf);
- shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.remove(ShuffleServerConf.NETTY_SERVER_PORT.key());
- createShuffleServer(shuffleServerConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+
+ storeShuffleServerConf(shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(shuffleServerConfWithoutPort(1, tmpDir, ServerType.GRPC));
+
+ startServersWithRandomPorts();
}
@Test
@@ -141,31 +138,28 @@ public class CoordinatorGrpcTest extends CoordinatorTestBase {
}
@Test
- public void getShuffleAssignmentsTest() throws Exception {
+ public void getShuffleAssignmentsTest(@TempDir File tmpDir) throws Exception {
final String appId = "getShuffleAssignmentsTest";
CoordinatorTestUtils.waitForRegister(coordinatorClient, 2);
- // When the shuffleServerHeartbeat Test is completed before the current test,
- // the server's tags will be [ss_v4, GRPC_NETTY] and [ss_v4, GRPC], respectively.
- // We need to remove the first machine's tag from GRPC_NETTY to GRPC
+
grpcShuffleServers.get(0).stopServer();
- RssConf shuffleServerConf = grpcShuffleServers.get(0).getShuffleServerConf();
- Class<RssConf> clazz = RssConf.class;
- Field field = clazz.getDeclaredField("settings");
- field.setAccessible(true);
- ((ConcurrentHashMap<Object, Object>) field.get(shuffleServerConf))
- .remove(ShuffleServerConf.NETTY_SERVER_PORT.key());
+ List<Integer> ports = reserveJettyPorts(1);
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC);
+ shuffleServerConf.set(ShuffleServerConf.STORAGE_MEDIA_PROVIDER_ENV_KEY, "RSS_ENV_KEY");
+ String baseDir = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH).get(0);
+ shuffleServerConf.setString(ShuffleServerConf.RSS_STORAGE_BASE_PATH.key(), baseDir);
String storageTypeJsonSource = String.format("{\"%s\": \"ssd\"}", baseDir);
-
withEnvironmentVariables("RSS_ENV_KEY", storageTypeJsonSource)
.execute(
() -> {
- ShuffleServerConf tempShuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- tempShuffleServerConf.remove(ShuffleServerConf.NETTY_SERVER_PORT.key());
- ShuffleServer ss = new ShuffleServer(tempShuffleServerConf);
+ shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
+ shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
+ shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
+ ShuffleServer ss = new ShuffleServer(shuffleServerConf);
ss.start();
grpcShuffleServers.set(0, ss);
});
- Thread.sleep(5000);
+ Thread.sleep(4000);
CoordinatorServer coordinatorServer = coordinators.get(0);
((SimpleClusterManager) (coordinatorServer.getClusterManager())).nodesCheckTest();
@@ -287,7 +281,7 @@ public class CoordinatorGrpcTest extends CoordinatorTestBase {
}
@Test
- public void shuffleServerHeartbeatTest() throws Exception {
+ public void shuffleServerHeartbeatTest(@TempDir File tempDir) throws Exception {
CoordinatorTestUtils.waitForRegister(coordinatorClient, 2);
grpcShuffleServers.get(0).stopServer();
Thread.sleep(5000);
@@ -303,15 +297,21 @@ public class CoordinatorGrpcTest extends CoordinatorTestBase {
assertEquals(StorageStatus.NORMAL, infoHead.getStatus());
assertTrue(node.getTags().contains(Constants.SHUFFLE_SERVER_VERSION));
assertTrue(scm.getTagToNodes().get(Constants.SHUFFLE_SERVER_VERSION).contains(node));
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
+
+ List<Integer> ports = reserveJettyPorts(1);
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tempDir, ServerType.GRPC);
shuffleServerConf.set(ShuffleServerConf.STORAGE_MEDIA_PROVIDER_ENV_KEY, "RSS_ENV_KEY");
String baseDir = shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_BASE_PATH).get(0);
+ shuffleServerConf.setString(ShuffleServerConf.RSS_STORAGE_BASE_PATH.key(), baseDir);
String storageTypeJsonSource = String.format("{\"%s\": \"ssd\"}", baseDir);
withEnvironmentVariables("RSS_ENV_KEY", storageTypeJsonSource)
.execute(
() -> {
// set this server's tag to ssd
shuffleServerConf.set(ShuffleServerConf.TAGS, Lists.newArrayList("SSD"));
+ shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
+ shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
+ shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
ShuffleServer ss = new ShuffleServer(shuffleServerConf);
ss.start();
grpcShuffleServers.set(0, ss);
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorReconfigureNodeMaxTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorReconfigureNodeMaxTest.java
index 4eb10caa0..57004b755 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorReconfigureNodeMaxTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorReconfigureNodeMaxTest.java
@@ -20,7 +20,6 @@ package org.apache.uniffle.test;
import java.io.File;
import java.io.FileWriter;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
@@ -49,48 +48,38 @@ public class CoordinatorReconfigureNodeMaxTest extends CoordinatorTestBase {
private static final int SERVER_NUM = 10;
private static final HashSet<String> TAGS = Sets.newHashSet("t1");
- private static final String QUORUM =
- LOCALHOST + ":" + COORDINATOR_PORT_1 + "," + LOCALHOST + ":" + COORDINATOR_PORT_2;
-
private static String tempConfFilePath;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf1 = getCoordinatorConf();
+ CoordinatorConf coordinatorConf1 = coordinatorConfWithoutPort();
coordinatorConf1.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
coordinatorConf1.setInteger(
CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, DEFAULT_SHUFFLE_NODES_MAX);
coordinatorConf1.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
tempConfFilePath = tmpDir.getPath() + "/conf.file";
ReconfigurableConfManager.init(coordinatorConf1, tempConfFilePath);
- createCoordinatorServer(coordinatorConf1);
+ storeCoordinatorConf(coordinatorConf1);
- CoordinatorConf coordinatorConf2 = getCoordinatorConf();
+ CoordinatorConf coordinatorConf2 = coordinatorConfWithoutPort();
coordinatorConf2.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
coordinatorConf2.setInteger(
CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, DEFAULT_SHUFFLE_NODES_MAX);
- coordinatorConf2.setInteger(CoordinatorConf.RPC_SERVER_PORT, COORDINATOR_PORT_2);
- coordinatorConf2.setInteger(CoordinatorConf.JETTY_HTTP_PORT, JETTY_PORT_2);
coordinatorConf2.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
- createCoordinatorServer(coordinatorConf2);
+ storeCoordinatorConf(coordinatorConf2);
for (int i = 0; i < SERVER_NUM; i++) {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- File dataDir = new File(tmpDir, "data" + i);
- String basePath = dataDir.getAbsolutePath();
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(i, tmpDir, ServerType.GRPC);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.MEMORY_LOCALFILE_HDFS.name());
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 2000L);
shuffleServerConf.set(ShuffleServerConf.SERVER_PRE_ALLOCATION_EXPIRED, 5000L);
- shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 18001 + i);
- shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, 19010 + i);
shuffleServerConf.set(ShuffleServerConf.TAGS, new ArrayList<>(TAGS));
- shuffleServerConf.setString("rss.coordinator.quorum", QUORUM);
- createShuffleServer(shuffleServerConf);
+ storeShuffleServerConf(shuffleServerConf);
}
- startServers();
+ startServersWithRandomPorts();
Thread.sleep(1000 * 5);
}
@@ -117,7 +106,7 @@ public class CoordinatorReconfigureNodeMaxTest extends CoordinatorTestBase {
.unregisterThreadPoolSize(10)
.unregisterRequestTimeSec(10)
.build();
- shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
+ shuffleWriteClient.registerCoordinators(getQuorum());
ShuffleAssignmentsInfo info =
shuffleWriteClient.getShuffleAssignments("app1", 0, 10, 1, TAGS, SERVER_NUM + 10, -1);
assertEquals(DEFAULT_SHUFFLE_NODES_MAX, info.getServerToPartitionRanges().keySet().size());
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorTestBase.java b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorTestBase.java
index 4c6def032..e09c5f394 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorTestBase.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorTestBase.java
@@ -29,11 +29,17 @@ public class CoordinatorTestBase extends IntegrationTestBase {
protected CoordinatorClientFactory factory = CoordinatorClientFactory.getInstance();
protected CoordinatorGrpcClient coordinatorClient;
+ protected CoordinatorGrpcClient getCoordinatorClient() {
+ return (CoordinatorGrpcClient)
+ factory.createCoordinatorClient(
+ ClientType.GRPC, LOCALHOST, coordinators.get(0).getRpcListenPort());
+ }
+
@BeforeEach
public void createClient() {
- coordinatorClient =
- (CoordinatorGrpcClient)
- factory.createCoordinatorClient(ClientType.GRPC, LOCALHOST, COORDINATOR_PORT_1);
+ if (!coordinators.isEmpty()) {
+ coordinatorClient = getCoordinatorClient();
+ }
}
@AfterEach
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/DiskErrorToleranceTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/DiskErrorToleranceTest.java
index 9cfe686e5..db1962e3e 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/DiskErrorToleranceTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/DiskErrorToleranceTest.java
@@ -53,7 +53,6 @@ import org.apache.uniffle.common.ShuffleBlockInfo;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.common.rpc.StatusCode;
-import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
@@ -63,8 +62,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class DiskErrorToleranceTest extends ShuffleReadWriteBase {
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
private static File data1;
private static File data2;
@@ -75,52 +72,37 @@ public class DiskErrorToleranceTest extends ShuffleReadWriteBase {
public static void setupServers(@TempDir File serverTmpDir) throws Exception {
data1 = new File(serverTmpDir, "data1");
data2 = new File(serverTmpDir, "data2");
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ storeCoordinatorConf(coordinatorConfWithoutPort());
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC);
- createShuffleServer(grpcShuffleServerConf);
+ storeShuffleServerConf(buildShuffleServerConf(ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(ServerType.GRPC_NETTY));
- ShuffleServerConf nettyShuffleServerConf = buildShuffleServerConf(ServerType.GRPC_NETTY);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
@BeforeEach
public void createClient(@TempDir File serverTmpDir) throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
grpcShuffleServerInfoList =
Lists.newArrayList(
- new ShuffleServerInfo(
- String.format(
- "127.0.0.1-%s",
- grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT)),
- LOCALHOST,
- grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT)));
+ new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort()));
nettyShuffleServerInfoList =
Lists.newArrayList(
new ShuffleServerInfo(
- String.format(
- "127.0.0.1-%s",
- grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT)),
LOCALHOST,
- grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT)));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort()));
}
private static ShuffleServerConf buildShuffleServerConf(ServerType serverType) throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, null, serverType);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE.name());
shuffleServerConf.set(
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/FetchClientConfTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/FetchClientConfTest.java
index 01d2238b1..72d6aa3b2 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/FetchClientConfTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/FetchClientConfTest.java
@@ -30,6 +30,7 @@ import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.apache.uniffle.client.impl.grpc.CoordinatorGrpcClient;
import org.apache.uniffle.client.request.RssFetchClientConfRequest;
import org.apache.uniffle.client.request.RssFetchRemoteStorageRequest;
import org.apache.uniffle.client.response.RssFetchClientConfResponse;
@@ -56,34 +57,37 @@ public class FetchClientConfTest extends CoordinatorTestBase {
printWriter.flush();
printWriter.close();
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setBoolean(CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED, true);
coordinatorConf.setString(
CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_PATH, clientConfFile.getAbsolutePath());
coordinatorConf.setInteger("rss.coordinator.dynamicClientConf.updateIntervalSec", 10);
- createCoordinatorServer(coordinatorConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+ startServersWithRandomPorts();
Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
RssFetchClientConfRequest request = new RssFetchClientConfRequest(2000);
- RssFetchClientConfResponse response = coordinatorClient.fetchClientConf(request);
+ CoordinatorGrpcClient client1 = getCoordinatorClient();
+ RssFetchClientConfResponse response = client1.fetchClientConf(request);
assertEquals(StatusCode.SUCCESS, response.getStatusCode());
assertEquals(2, response.getClientConf().size());
assertEquals("1234", response.getClientConf().get("spark.mock.1"));
assertEquals("true", response.getClientConf().get("spark.mock.2"));
assertNull(response.getClientConf().get("spark.mock.3"));
+ client1.close();
shutdownServers();
// dynamic client conf is disabled by default
- coordinatorConf = getCoordinatorConf();
+ coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setString(
"rss.coordinator.dynamicClientConf.path", clientConfFile.getAbsolutePath());
coordinatorConf.setInteger("rss.coordinator.dynamicClientConf.updateIntervalSec", 10);
- createCoordinatorServer(coordinatorConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+ startServersWithRandomPorts();
Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
request = new RssFetchClientConfRequest(2000);
- response = coordinatorClient.fetchClientConf(request);
+ CoordinatorGrpcClient client2 = getCoordinatorClient();
+ response = client2.fetchClientConf(request);
assertEquals(StatusCode.SUCCESS, response.getStatusCode());
assertEquals(0, response.getClientConf().size());
shutdownServers();
@@ -91,10 +95,10 @@ public class FetchClientConfTest extends CoordinatorTestBase {
// Fetch conf will not throw exception even if the request fails
Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
request = new RssFetchClientConfRequest(10);
- response = coordinatorClient.fetchClientConf(request);
+ response = client2.fetchClientConf(request);
assertEquals(StatusCode.INTERNAL_ERROR, response.getStatusCode());
assertEquals(0, response.getClientConf().size());
- shutdownServers();
+ client2.close();
}
@Test
@@ -107,7 +111,7 @@ public class FetchClientConfTest extends CoordinatorTestBase {
dynamicConf.put(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_CLUSTER_CONF.key(), contItem);
writeRemoteStorageConf(cfgFile, dynamicConf);
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setBoolean(CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED, true);
coordinatorConf.setString(
CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_PATH, cfgFile.toURI().toString());
@@ -115,13 +119,14 @@ public class FetchClientConfTest extends CoordinatorTestBase {
CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_UPDATE_INTERVAL_SEC, 3);
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_TIME, 1000);
coordinatorConf.setInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_ACCESS_TIMES, 1);
- createCoordinatorServer(coordinatorConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+ startServersWithRandomPorts();
waitForUpdate(Sets.newHashSet(remotePath1), coordinators.get(0).getApplicationManager());
String appId = "application_testFetchRemoteStorageApp_" + 1;
RssFetchRemoteStorageRequest request = new RssFetchRemoteStorageRequest(appId);
- RssFetchRemoteStorageResponse response = coordinatorClient.fetchRemoteStorage(request);
+ CoordinatorGrpcClient client1 = getCoordinatorClient();
+ RssFetchRemoteStorageResponse response = client1.fetchRemoteStorage(request);
RemoteStorageInfo remoteStorageInfo = response.getRemoteStorageInfo();
assertTrue(remoteStorageInfo.getConfItems().isEmpty());
assertEquals(remotePath1, remoteStorageInfo.getPath());
@@ -133,7 +138,7 @@ public class FetchClientConfTest extends CoordinatorTestBase {
waitForUpdate(Sets.newHashSet(remotePath2), coordinators.get(0).getApplicationManager());
request = new RssFetchRemoteStorageRequest(appId);
Thread.sleep(1500);
- response = coordinatorClient.fetchRemoteStorage(request);
+ response = client1.fetchRemoteStorage(request);
// remotePath1 will be return because (appId -> remote storage path) is in cache
remoteStorageInfo = response.getRemoteStorageInfo();
assertEquals(remotePath1, remoteStorageInfo.getPath());
@@ -141,13 +146,14 @@ public class FetchClientConfTest extends CoordinatorTestBase {
String newAppId = "application_testFetchRemoteStorageApp_" + 2;
request = new RssFetchRemoteStorageRequest(newAppId);
- response = coordinatorClient.fetchRemoteStorage(request);
+ response = client1.fetchRemoteStorage(request);
// got the remotePath2 for new appId
remoteStorageInfo = response.getRemoteStorageInfo();
assertEquals(remotePath2, remoteStorageInfo.getPath());
assertEquals(2, remoteStorageInfo.getConfItems().size());
assertEquals("test1", remoteStorageInfo.getConfItems().get("key1"));
assertEquals("test2", remoteStorageInfo.getConfItems().get("key2"));
+ client1.close();
shutdownServers();
}
@@ -161,7 +167,7 @@ public class FetchClientConfTest extends CoordinatorTestBase {
dynamicConf.put(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_CLUSTER_CONF.key(), contItem);
writeRemoteStorageConf(cfgFile, dynamicConf);
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setBoolean(CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED, true);
coordinatorConf.setString(
CoordinatorConf.COORDINATOR_DYNAMIC_CLIENT_CONF_PATH, cfgFile.toURI().toString());
@@ -170,13 +176,14 @@ public class FetchClientConfTest extends CoordinatorTestBase {
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_TIME, 1000);
coordinatorConf.setInteger(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SCHEDULE_ACCESS_TIMES, 1);
coordinatorConf.set(CoordinatorConf.COORDINATOR_REMOTE_STORAGE_SELECT_STRATEGY, IO_SAMPLE);
- createCoordinatorServer(coordinatorConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+ startServersWithRandomPorts();
waitForUpdate(Sets.newHashSet(remotePath1), coordinators.get(0).getApplicationManager());
String appId = "application_testFetchRemoteStorageApp_" + 1;
RssFetchRemoteStorageRequest request = new RssFetchRemoteStorageRequest(appId);
- RssFetchRemoteStorageResponse response = coordinatorClient.fetchRemoteStorage(request);
+ CoordinatorGrpcClient client1 = getCoordinatorClient();
+ RssFetchRemoteStorageResponse response = client1.fetchRemoteStorage(request);
RemoteStorageInfo remoteStorageInfo = response.getRemoteStorageInfo();
assertTrue(remoteStorageInfo.getConfItems().isEmpty());
assertEquals(remotePath1, remoteStorageInfo.getPath());
@@ -187,7 +194,7 @@ public class FetchClientConfTest extends CoordinatorTestBase {
writeRemoteStorageConf(cfgFile, dynamicConf);
waitForUpdate(Sets.newHashSet(remotePath2), coordinators.get(0).getApplicationManager());
request = new RssFetchRemoteStorageRequest(appId);
- response = coordinatorClient.fetchRemoteStorage(request);
+ response = client1.fetchRemoteStorage(request);
// remotePath1 will be return because (appId -> remote storage path) is in cache
remoteStorageInfo = response.getRemoteStorageInfo();
assertEquals(remotePath1, remoteStorageInfo.getPath());
@@ -197,13 +204,14 @@ public class FetchClientConfTest extends CoordinatorTestBase {
Thread.sleep(2000);
String newAppId = "application_testFetchRemoteStorageApp_" + 2;
request = new RssFetchRemoteStorageRequest(newAppId);
- response = coordinatorClient.fetchRemoteStorage(request);
+ response = client1.fetchRemoteStorage(request);
// got the remotePath2 for new appId
remoteStorageInfo = response.getRemoteStorageInfo();
assertEquals(remotePath2, remoteStorageInfo.getPath());
assertEquals(2, remoteStorageInfo.getConfItems().size());
assertEquals("test1", remoteStorageInfo.getConfItems().get("key1"));
assertEquals("test2", remoteStorageInfo.getConfItems().get("key2"));
+ client1.close();
shutdownServers();
}
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckCoordinatorGrpcTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckCoordinatorGrpcTest.java
index 08202e3d7..6201c37f6 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckCoordinatorGrpcTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckCoordinatorGrpcTest.java
@@ -69,38 +69,38 @@ public class HealthCheckCoordinatorGrpcTest extends CoordinatorTestBase {
long usedSize = serverTmpDir.getTotalSpace() - serverTmpDir.getUsableSpace();
maxUsage = (writeDataSize * 0.75 + usedSize) * 100.0 / totalSize;
healthUsage = (writeDataSize * 0.5 + usedSize) * 100.0 / totalSize;
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_HEARTBEAT_TIMEOUT, 3000);
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.setBoolean(ShuffleServerConf.HEALTH_CHECK_ENABLE, true);
- shuffleServerConf.setString(
+ storeCoordinatorConf(coordinatorConf);
+
+ ShuffleServerConf shuffleServerConf1 =
+ shuffleServerConfWithoutPort(0, serverTmpDir, ServerType.GRPC);
+ shuffleServerConf1.setBoolean(ShuffleServerConf.HEALTH_CHECK_ENABLE, true);
+ shuffleServerConf1.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE.name());
- shuffleServerConf.set(
+ shuffleServerConf1.set(
ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(data1.getAbsolutePath()));
- shuffleServerConf.setDouble(
+ shuffleServerConf1.setDouble(
ShuffleServerConf.HEALTH_STORAGE_RECOVERY_USAGE_PERCENTAGE, healthUsage);
- shuffleServerConf.setDouble(ShuffleServerConf.HEALTH_STORAGE_MAX_USAGE_PERCENTAGE, maxUsage);
- shuffleServerConf.setLong(ShuffleServerConf.HEALTH_CHECK_INTERVAL, 1000L);
- createShuffleServer(shuffleServerConf);
- shuffleServerConf.setInteger(
- ShuffleServerConf.RPC_SERVER_PORT,
- shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT) + 1);
- shuffleServerConf.setInteger(
- ShuffleServerConf.JETTY_HTTP_PORT,
- shuffleServerConf.getInteger(ShuffleServerConf.JETTY_HTTP_PORT) + 1);
- shuffleServerConf.setString(
+ shuffleServerConf1.setDouble(ShuffleServerConf.HEALTH_STORAGE_MAX_USAGE_PERCENTAGE, maxUsage);
+ shuffleServerConf1.setLong(ShuffleServerConf.HEALTH_CHECK_INTERVAL, 1000L);
+ storeShuffleServerConf(shuffleServerConf1);
+
+ ShuffleServerConf shuffleServerConf2 =
+ shuffleServerConfWithoutPort(1, serverTmpDir, ServerType.GRPC);
+ shuffleServerConf2.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE.name());
- shuffleServerConf.set(
+ shuffleServerConf2.set(
ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(data2.getAbsolutePath()));
- shuffleServerConf.setDouble(
+ shuffleServerConf2.setDouble(
ShuffleServerConf.HEALTH_STORAGE_RECOVERY_USAGE_PERCENTAGE, healthUsage);
- shuffleServerConf.setDouble(ShuffleServerConf.HEALTH_STORAGE_MAX_USAGE_PERCENTAGE, maxUsage);
- shuffleServerConf.setLong(ShuffleServerConf.HEALTH_CHECK_INTERVAL, 1000L);
- shuffleServerConf.setBoolean(ShuffleServerConf.HEALTH_CHECK_ENABLE, true);
- createShuffleServer(shuffleServerConf);
- startServers();
+ shuffleServerConf2.setDouble(ShuffleServerConf.HEALTH_STORAGE_MAX_USAGE_PERCENTAGE, maxUsage);
+ shuffleServerConf2.setLong(ShuffleServerConf.HEALTH_CHECK_INTERVAL, 1000L);
+ shuffleServerConf2.setBoolean(ShuffleServerConf.HEALTH_CHECK_ENABLE, true);
+ storeShuffleServerConf(shuffleServerConf2);
+
+ startServersWithRandomPorts();
}
@Test
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckTest.java
index 56173cbd0..3cc198fb5 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/HealthCheckTest.java
@@ -37,7 +37,7 @@ import org.apache.uniffle.storage.util.StorageType;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public class HealthCheckTest extends CoordinatorTestBase {
+public class HealthCheckTest {
@BeforeAll
public static void setup() {
ShuffleServerMetrics.register();
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageFaultToleranceBase.java b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageFaultToleranceBase.java
index 8a2fe2635..0b916ec51 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageFaultToleranceBase.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageFaultToleranceBase.java
@@ -51,7 +51,6 @@ import org.apache.uniffle.common.PartitionRange;
import org.apache.uniffle.common.ShuffleBlockInfo;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.rpc.StatusCode;
-import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -60,21 +59,18 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public abstract class HybridStorageFaultToleranceBase extends ShuffleReadWriteBase {
protected ShuffleServerGrpcClient grpcShuffleServerClient;
protected ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- protected static ShuffleServerConf grpcShuffleServerConfig;
- protected static ShuffleServerConf nettyShuffleServerConfig;
private static String REMOTE_STORAGE = HDFS_URI + "rss/multi_storage_fault_%s";
@BeforeEach
public void createClient() throws Exception {
ShuffleServerClientFactory.getInstance().cleanupCache();
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
@@ -164,10 +160,9 @@ public abstract class HybridStorageFaultToleranceBase extends ShuffleReadWriteBa
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
ShuffleReadClientImpl readClient =
ShuffleClientFactory.newReadBuilder()
.clientType(isNettyMode ? ClientType.GRPC_NETTY : ClientType.GRPC)
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageHadoopFallbackTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageHadoopFallbackTest.java
index 4ec4fff32..937a38203 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageHadoopFallbackTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageHadoopFallbackTest.java
@@ -18,17 +18,11 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.Arrays;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;
-import org.apache.uniffle.client.factory.ShuffleServerClientFactory;
-import org.apache.uniffle.client.impl.grpc.ShuffleServerGrpcClient;
-import org.apache.uniffle.client.impl.grpc.ShuffleServerGrpcNettyClient;
import org.apache.uniffle.common.rpc.ServerType;
-import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
@@ -38,27 +32,18 @@ public class HybridStorageHadoopFallbackTest extends HybridStorageFaultTolerance
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- final CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ storeCoordinatorConf(coordinatorConfWithoutPort());
- String basePath = generateBasePath(tmpDir);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC_NETTY));
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, basePath);
- createShuffleServer(grpcShuffleServerConf);
-
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, basePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tmpDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tmpDir, serverType);
shuffleServerConf.setDouble(ShuffleServerConf.CLEANUP_THRESHOLD, 0.0);
shuffleServerConf.setDouble(ShuffleServerConf.HIGH_WATER_MARK_OF_WRITE, 100.0);
shuffleServerConf.setLong(ShuffleServerConf.DISK_CAPACITY, 1024L * 1024L * 100);
@@ -68,25 +53,11 @@ public class HybridStorageHadoopFallbackTest extends HybridStorageFaultTolerance
shuffleServerConf.setLong(ShuffleServerConf.SERVER_COMMIT_TIMEOUT, 20L * 1000L);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE_HDFS.name());
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.setLong(
ShuffleServerConf.FLUSH_COLD_STORAGE_THRESHOLD_SIZE, 1L * 1024L * 1024L);
return shuffleServerConf;
}
- @BeforeEach
- public void createClient() throws Exception {
- ShuffleServerClientFactory.getInstance().cleanupCache();
- grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
- nettyShuffleServerClient =
- new ShuffleServerGrpcNettyClient(
- LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
- }
-
@Override
public void makeChaos() {
assertEquals(1, cluster.getDataNodes().size());
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageLocalFileFallbackTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageLocalFileFallbackTest.java
index 15680573b..a574de91a 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageLocalFileFallbackTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/HybridStorageLocalFileFallbackTest.java
@@ -18,7 +18,6 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.Arrays;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -41,40 +40,31 @@ public class HybridStorageLocalFileFallbackTest extends HybridStorageFaultTolera
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- final CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ final CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- String basePath = generateBasePath(tmpDir);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC_NETTY));
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, basePath);
- createShuffleServer(grpcShuffleServerConf);
-
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, basePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
@BeforeEach
public void createClient() throws Exception {
ShuffleServerClientFactory.getInstance().cleanupCache();
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tmpDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tmpDir, serverType);
shuffleServerConf.setDouble(ShuffleServerConf.CLEANUP_THRESHOLD, 0.0);
shuffleServerConf.setDouble(ShuffleServerConf.HIGH_WATER_MARK_OF_WRITE, 100.0);
shuffleServerConf.setLong(ShuffleServerConf.DISK_CAPACITY, 1024L * 1024L * 100);
@@ -84,7 +74,6 @@ public class HybridStorageLocalFileFallbackTest extends HybridStorageFaultTolera
shuffleServerConf.setLong(ShuffleServerConf.SERVER_COMMIT_TIMEOUT, 20L * 1000L);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE_HDFS.name());
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.setLong(
ShuffleServerConf.FLUSH_COLD_STORAGE_THRESHOLD_SIZE, 1000L * 1024L * 1024L);
shuffleServerConf.setString(
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/IntegrationTestBase.java b/integration-test/common/src/test/java/org/apache/uniffle/test/IntegrationTestBase.java
index bd827616c..2e89fa942 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/IntegrationTestBase.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/IntegrationTestBase.java
@@ -21,12 +21,16 @@ import java.io.File;
import java.io.FileWriter;
import java.io.PrintWriter;
import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.io.TempDir;
@@ -97,6 +101,32 @@ public abstract class IntegrationTestBase extends HadoopTestBase {
}
}
+ public static String getQuorum() {
+ return coordinators.stream()
+ .map(CoordinatorServer::getRpcListenPort)
+ .map(port -> LOCALHOST + ":" + port)
+ .collect(Collectors.joining(","));
+ }
+
+ public static List<Integer> generateNonExistingPorts(int num) {
+ Set<Integer> portExistsSet = Sets.newHashSet();
+ jettyPorts.forEach(port -> portExistsSet.add(port));
+ coordinators.forEach(server -> portExistsSet.add(server.getRpcListenPort()));
+ grpcShuffleServers.forEach(server -> portExistsSet.add(server.getGrpcPort()));
+ nettyShuffleServers.forEach(server -> portExistsSet.add(server.getGrpcPort()));
+ nettyShuffleServers.forEach(server -> portExistsSet.add(server.getNettyPort()));
+ int i = 0;
+ List<Integer> fakePorts = new ArrayList<>(num);
+ while (i < num) {
+ int port = ThreadLocalRandom.current().nextInt(1024, 65535);
+ if (portExistsSet.add(port)) {
+ fakePorts.add(port);
+ i++;
+ }
+ }
+ return fakePorts;
+ }
+
public static void startServersWithRandomPorts() throws Exception {
final int jettyPortSize =
coordinatorConfList.size()
@@ -113,11 +143,7 @@ public abstract class IntegrationTestBase extends HadoopTestBase {
for (CoordinatorServer coordinator : coordinators) {
coordinator.start();
}
- String quorum =
- coordinators.stream()
- .map(CoordinatorServer::getRpcListenPort)
- .map(port -> LOCALHOST + ":" + port)
- .collect(Collectors.joining(","));
+ String quorum = getQuorum();
for (ShuffleServerConf serverConf : shuffleServerConfList) {
serverConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, jettyPorts.get(index));
@@ -144,10 +170,14 @@ public abstract class IntegrationTestBase extends HadoopTestBase {
protected static List<Integer> jettyPorts = Lists.newArrayList();
- public static void reserveJettyPorts(int numPorts) {
+ public static List<Integer> reserveJettyPorts(int numPorts) {
+ List<Integer> ports = new ArrayList<>(numPorts);
for (int i = 0; i < numPorts; i++) {
- jettyPorts.add(PortRegistry.reservePort());
+ int port = PortRegistry.reservePort();
+ jettyPorts.add(port);
+ ports.add(port);
}
+ return ports;
}
@AfterAll
@@ -242,10 +272,12 @@ public abstract class IntegrationTestBase extends HadoopTestBase {
protected static ShuffleServerConf shuffleServerConfWithoutPort(
int subDirIndex, File tmpDir, ServerType serverType) {
ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType, "", 0, 0, 0);
- File dataDir1 = new File(tmpDir, subDirIndex + "_1");
- File dataDir2 = new File(tmpDir, subDirIndex + "_2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- shuffleServerConf.setString("rss.storage.basePath", basePath);
+ if (tmpDir != null) {
+ File dataDir1 = new File(tmpDir, subDirIndex + "_1");
+ File dataDir2 = new File(tmpDir, subDirIndex + "_2");
+ String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
+ shuffleServerConf.setString("rss.storage.basePath", basePath);
+ }
return shuffleServerConf;
}
@@ -305,13 +337,6 @@ public abstract class IntegrationTestBase extends HadoopTestBase {
}
}
- protected static void createAndStartServers(
- ShuffleServerConf shuffleServerConf, CoordinatorConf coordinatorConf) throws Exception {
- createCoordinatorServer(coordinatorConf);
- createShuffleServer(shuffleServerConf);
- startServers();
- }
-
protected static File createDynamicConfFile(Map<String, String> dynamicConf) throws Exception {
File dynamicConfFile = Files.createTempFile("dynamicConf", "conf").toFile();
writeRemoteStorageConf(dynamicConfFile, dynamicConf);
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/PartitionBalanceCoordinatorGrpcTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/PartitionBalanceCoordinatorGrpcTest.java
index 9884fcb13..4513de744 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/PartitionBalanceCoordinatorGrpcTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/PartitionBalanceCoordinatorGrpcTest.java
@@ -17,12 +17,14 @@
package org.apache.uniffle.test;
+import java.io.File;
import java.util.List;
import java.util.Map;
import com.google.common.collect.Sets;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.apache.uniffle.client.request.RssGetShuffleAssignmentsRequest;
import org.apache.uniffle.client.response.RssGetShuffleAssignmentsResponse;
@@ -36,35 +38,26 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class PartitionBalanceCoordinatorGrpcTest extends CoordinatorTestBase {
- private static int rpcPort1;
- private static int rpcPort2;
- private static int rpcPort3;
-
@BeforeAll
- public static void setupServers() throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ public static void setupServers(@TempDir File tmpDir) throws Exception {
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setLong("rss.coordinator.app.expired", 2000);
coordinatorConf.setLong("rss.coordinator.server.heartbeat.timeout", 30000);
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.setString("rss.server.buffer.capacity", "204800000");
- rpcPort1 = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
- shuffleServerConf.setInteger(
- "rss.rpc.server.port", shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT) + 1);
- shuffleServerConf.setInteger(
- "rss.jetty.http.port", shuffleServerConf.getInteger(ShuffleServerConf.JETTY_HTTP_PORT) + 1);
- shuffleServerConf.setString("rss.server.buffer.capacity", "512000000");
- rpcPort2 = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
- shuffleServerConf.setInteger(
- "rss.rpc.server.port", shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT) + 2);
- shuffleServerConf.setInteger(
- "rss.jetty.http.port", shuffleServerConf.getInteger(ShuffleServerConf.JETTY_HTTP_PORT) + 2);
- shuffleServerConf.setString("rss.server.buffer.capacity", "102400000");
- rpcPort3 = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
- startServers();
+ storeCoordinatorConf(coordinatorConf);
+
+ ShuffleServerConf shuffleServerConf0 = shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC);
+ shuffleServerConf0.setString("rss.server.buffer.capacity", "204800000");
+ storeShuffleServerConf(shuffleServerConf0);
+
+ ShuffleServerConf shuffleServerConf1 = shuffleServerConfWithoutPort(1, tmpDir, ServerType.GRPC);
+ shuffleServerConf1.setString("rss.server.buffer.capacity", "512000000");
+ storeShuffleServerConf(shuffleServerConf1);
+
+ ShuffleServerConf shuffleServerConf2 = shuffleServerConfWithoutPort(2, tmpDir, ServerType.GRPC);
+ shuffleServerConf2.setString("rss.server.buffer.capacity", "102400000");
+ storeShuffleServerConf(shuffleServerConf2);
+
+ startServersWithRandomPorts();
}
@Test
@@ -78,7 +71,7 @@ public class PartitionBalanceCoordinatorGrpcTest extends CoordinatorTestBase {
for (Map.Entry<Integer, List<ShuffleServerInfo>> entry :
response.getPartitionToServers().entrySet()) {
assertEquals(1, entry.getValue().size());
- assertEquals(rpcPort2, entry.getValue().get(0).getGrpcPort());
+ assertEquals(grpcShuffleServers.get(1).getGrpcPort(), entry.getValue().get(0).getGrpcPort());
}
request =
new RssGetShuffleAssignmentsRequest(
@@ -88,7 +81,7 @@ public class PartitionBalanceCoordinatorGrpcTest extends CoordinatorTestBase {
for (Map.Entry<Integer, List<ShuffleServerInfo>> entry :
response.getPartitionToServers().entrySet()) {
assertEquals(1, entry.getValue().size());
- assertEquals(rpcPort2, entry.getValue().get(0).getGrpcPort());
+ assertEquals(grpcShuffleServers.get(1).getGrpcPort(), entry.getValue().get(0).getGrpcPort());
}
request =
new RssGetShuffleAssignmentsRequest(
@@ -98,7 +91,7 @@ public class PartitionBalanceCoordinatorGrpcTest extends CoordinatorTestBase {
for (Map.Entry<Integer, List<ShuffleServerInfo>> entry :
response.getPartitionToServers().entrySet()) {
assertEquals(1, entry.getValue().size());
- assertEquals(rpcPort1, entry.getValue().get(0).getGrpcPort());
+ assertEquals(grpcShuffleServers.get(0).getGrpcPort(), entry.getValue().get(0).getGrpcPort());
}
}
}
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/QuorumTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/QuorumTest.java
index 9be5ef78d..b321c4200 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/QuorumTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/QuorumTest.java
@@ -18,12 +18,9 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ThreadLocalRandom;
-import java.util.stream.Collectors;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -47,13 +44,12 @@ import org.apache.uniffle.common.RemoteStorageInfo;
import org.apache.uniffle.common.ShuffleBlockInfo;
import org.apache.uniffle.common.ShuffleDataDistributionType;
import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.config.RssBaseConf;
import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.common.util.RssUtils;
import org.apache.uniffle.coordinator.CoordinatorConf;
-import org.apache.uniffle.coordinator.CoordinatorServer;
import org.apache.uniffle.server.MockedGrpcServer;
import org.apache.uniffle.server.MockedShuffleServer;
-import org.apache.uniffle.server.ShuffleServer;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
@@ -88,92 +84,35 @@ public class QuorumTest extends ShuffleReadWriteBase {
.readBufferSize(1000);
}
- public static MockedShuffleServer createServer(int id, File tmpDir, int coordinatorRpcPort)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.setInteger("rss.rpc.server.port", 0);
+ public static ShuffleServerConf buildServerConf(int id, File tmpDir) {
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(id, tmpDir, ServerType.GRPC);
// app expires time should be greater than the client send data time which is 180s by default.
// (rpcTimeout*retryTimes)
// this can avoid get NO_REGISTER when second data to secondaryServer
shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 240 * 1000L);
shuffleServerConf.setLong("rss.server.heartbeat.interval", 5000L);
- File dataDir1 = new File(tmpDir, id + "_1");
- File dataDir2 = new File(tmpDir, id + "_2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
shuffleServerConf.setString("rss.storage.type", StorageType.MEMORY_LOCALFILE.name());
- shuffleServerConf.setInteger("rss.jetty.http.port", 0);
- shuffleServerConf.setString("rss.storage.basePath", basePath);
- shuffleServerConf.setString("rss.coordinator.quorum", LOCALHOST + ":" + coordinatorRpcPort);
- return new MockedShuffleServer(shuffleServerConf);
- }
-
- private List<Integer> generateFakePort(int num) {
- Set<Integer> portExistsSet =
- grpcShuffleServers.stream().map(ShuffleServer::getGrpcPort).collect(Collectors.toSet());
- int i = 0;
- List<Integer> fakePorts = new ArrayList<>(num);
- while (i < num) {
- int port = ThreadLocalRandom.current().nextInt(1, 65535);
- if (portExistsSet.add(port)) {
- fakePorts.add(port);
- i++;
- }
- }
- return fakePorts;
+ return shuffleServerConf;
}
@BeforeEach
public void initCluster(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- coordinatorConf.setInteger(CoordinatorConf.RPC_SERVER_PORT, 0);
- coordinatorConf.setInteger(CoordinatorConf.JETTY_HTTP_PORT, 0);
- createCoordinatorServer(coordinatorConf);
-
- for (CoordinatorServer coordinator : coordinators) {
- coordinator.start();
- }
-
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 8000);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- grpcShuffleServers.add(createServer(0, tmpDir, coordinators.get(0).getRpcListenPort()));
- grpcShuffleServers.add(createServer(1, tmpDir, coordinators.get(0).getRpcListenPort()));
- grpcShuffleServers.add(createServer(2, tmpDir, coordinators.get(0).getRpcListenPort()));
- grpcShuffleServers.add(createServer(3, tmpDir, coordinators.get(0).getRpcListenPort()));
- grpcShuffleServers.add(createServer(4, tmpDir, coordinators.get(0).getRpcListenPort()));
-
- for (ShuffleServer shuffleServer : grpcShuffleServers) {
- shuffleServer.start();
+ for (int i = 0; i < 5; i++) {
+ storeMockShuffleServerConf(buildServerConf(i, tmpDir));
}
+ startServersWithRandomPorts();
- shuffleServerInfo0 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(0).getGrpcPort()),
- grpcShuffleServers.get(0).getIp(),
- grpcShuffleServers.get(0).getGrpcPort());
- shuffleServerInfo1 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(1).getGrpcPort()),
- grpcShuffleServers.get(1).getIp(),
- grpcShuffleServers.get(1).getGrpcPort());
- shuffleServerInfo2 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(2).getGrpcPort()),
- grpcShuffleServers.get(2).getIp(),
- grpcShuffleServers.get(2).getGrpcPort());
- shuffleServerInfo3 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(3).getGrpcPort()),
- grpcShuffleServers.get(3).getIp(),
- grpcShuffleServers.get(3).getGrpcPort());
- shuffleServerInfo4 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(4).getGrpcPort()),
- grpcShuffleServers.get(4).getIp(),
- grpcShuffleServers.get(4).getGrpcPort());
+ shuffleServerInfo0 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
+ shuffleServerInfo1 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(1).getGrpcPort());
+ shuffleServerInfo2 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(2).getGrpcPort());
+ shuffleServerInfo3 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(3).getGrpcPort());
+ shuffleServerInfo4 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(4).getGrpcPort());
// simulator of failed servers
- List<Integer> fakePortList = generateFakePort(5);
+ List<Integer> fakePortList = generateNonExistingPorts(5);
fakedShuffleServerInfo0 =
new ShuffleServerInfo(
"127.0.0.1-20001", grpcShuffleServers.get(0).getIp(), fakePortList.get(0));
@@ -660,9 +599,16 @@ public class QuorumTest extends ShuffleReadWriteBase {
assertEquals(0, result.getFailedBlockIds().size());
assertEquals(blockIdBitmap, succBlockIdBitmap);
+ // tricky to reserve port, it'll be released after test
+ @SuppressWarnings("checkstyle:VariableDeclarationUsageDistance")
+ List<Integer> ports = reserveJettyPorts(2);
// when one server is restarted, getShuffleResult should success
grpcShuffleServers.get(1).stopServer();
- grpcShuffleServers.set(1, createServer(1, tmpDir, coordinators.get(0).getRpcListenPort()));
+ ShuffleServerConf shuffleServerConf1 = buildServerConf(5, tmpDir);
+ shuffleServerConf1.setString("rss.coordinator.quorum", getQuorum());
+ shuffleServerConf1.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
+ shuffleServerConf1.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(0));
+ grpcShuffleServers.set(1, new MockedShuffleServer(shuffleServerConf1));
grpcShuffleServers.get(1).start();
report =
shuffleWriteClientImpl.getShuffleResult(
@@ -675,7 +621,11 @@ public class QuorumTest extends ShuffleReadWriteBase {
// when two servers are restarted, getShuffleResult should fail
grpcShuffleServers.get(2).stopServer();
- grpcShuffleServers.set(2, createServer(2, tmpDir, coordinators.get(0).getRpcListenPort()));
+ ShuffleServerConf shuffleServerConf2 = buildServerConf(5, tmpDir);
+ shuffleServerConf2.setString("rss.coordinator.quorum", getQuorum());
+ shuffleServerConf2.setInteger(RssBaseConf.RPC_SERVER_PORT, 0);
+ shuffleServerConf2.setInteger(RssBaseConf.JETTY_HTTP_PORT, ports.get(1));
+ grpcShuffleServers.set(2, new MockedShuffleServer(shuffleServerConf2));
grpcShuffleServers.get(2).start();
try {
report =
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTest.java
index 760b27b8a..809e28a06 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTest.java
@@ -19,7 +19,6 @@ package org.apache.uniffle.test;
import java.io.File;
import java.io.IOException;
-import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
@@ -66,7 +65,6 @@ import org.apache.uniffle.common.util.BlockIdLayout;
import org.apache.uniffle.common.util.ChecksumUtils;
import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.proto.RssProtos;
-import org.apache.uniffle.server.ShuffleServer;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.server.buffer.ShuffleBufferType;
import org.apache.uniffle.storage.util.StorageType;
@@ -85,49 +83,26 @@ public class RemoteMergeShuffleWithRssClientTest extends ShuffleReadWriteBase {
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setBoolean(COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED, false);
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC_NETTY);
+ storeCoordinatorConf(coordinatorConf);
+
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC_NETTY);
shuffleServerConf.set(ShuffleServerConf.SERVER_MERGE_ENABLE, true);
shuffleServerConf.set(ShuffleServerConf.SERVER_MERGE_DEFAULT_MERGED_BLOCK_SIZE, "1k");
shuffleServerConf.set(
ShuffleServerConf.SERVER_SHUFFLE_BUFFER_TYPE, ShuffleBufferType.SKIP_LIST);
shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 10000000);
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
- List<Integer> ports = findAvailablePorts(2);
- shuffleServerConf.setInteger("rss.rpc.server.port", ports.get(0));
- shuffleServerConf.setInteger("rss.jetty.http.port", ports.get(1));
- createShuffleServer(shuffleServerConf);
- startServers();
- ShuffleServer shuffleServer = nettyShuffleServers.get(0);
+ storeShuffleServerConf(shuffleServerConf);
+
+ startServersWithRandomPorts();
shuffleServerInfo =
new ShuffleServerInfo(
- "127.0.0.1-20001",
- shuffleServer.getIp(),
- shuffleServer.getGrpcPort(),
- shuffleServer.getNettyPort());
- }
-
- private static List<Integer> findAvailablePorts(int num) throws IOException {
- List<ServerSocket> sockets = new ArrayList<>();
- List<Integer> ports = new ArrayList<>();
-
- for (int i = 0; i < num; i++) {
- ServerSocket socket = new ServerSocket(0);
- ports.add(socket.getLocalPort());
- sockets.add(socket);
- }
-
- for (ServerSocket socket : sockets) {
- socket.close();
- }
-
- return ports;
+ LOCALHOST,
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
public void createClient(String clientType) {
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed.java b/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed.java
index e63aee262..51276e128 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed.java
@@ -19,7 +19,6 @@ package org.apache.uniffle.test;
import java.io.File;
import java.io.IOException;
-import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
@@ -67,7 +66,6 @@ import org.apache.uniffle.common.util.BlockIdLayout;
import org.apache.uniffle.common.util.ChecksumUtils;
import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.proto.RssProtos;
-import org.apache.uniffle.server.ShuffleServer;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.server.buffer.ShuffleBufferType;
import org.apache.uniffle.server.storage.MultiPartLocalStorageManager;
@@ -90,15 +88,18 @@ public class RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed extends Shuff
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setBoolean(COORDINATOR_DYNAMIC_CLIENT_CONF_ENABLED, false);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC_NETTY);
+ storeCoordinatorConf(coordinatorConf);
+
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC_NETTY);
Assumptions.assumeTrue(
!shuffleServerConf
.get(SERVER_LOCAL_STORAGE_MANAGER_CLASS)
.equals(MultiPartLocalStorageManager.class.getName()),
MultiPartLocalStorageManager.class.getName() + " is not working with remote merge feature");
- createCoordinatorServer(coordinatorConf);
+
shuffleServerConf.set(ShuffleServerConf.SERVER_MERGE_ENABLE, true);
shuffleServerConf.set(ShuffleServerConf.SERVER_MERGE_DEFAULT_MERGED_BLOCK_SIZE, "1k");
shuffleServerConf.set(
@@ -107,40 +108,15 @@ public class RemoteMergeShuffleWithRssClientTestWhenShuffleFlushed extends Shuff
shuffleServerConf.set(SERVER_MEMORY_SHUFFLE_HIGHWATERMARK_PERCENTAGE, 0.0);
shuffleServerConf.set(SERVER_MEMORY_SHUFFLE_LOWWATERMARK_PERCENTAGE, 0.0);
shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 10000000);
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
- List<Integer> ports = findAvailablePorts(2);
- shuffleServerConf.setInteger("rss.rpc.server.port", ports.get(0));
- shuffleServerConf.setInteger("rss.jetty.http.port", ports.get(1));
- createShuffleServer(shuffleServerConf);
- startServers();
- ShuffleServer shuffleServer = nettyShuffleServers.get(0);
+ storeShuffleServerConf(shuffleServerConf);
+
+ startServersWithRandomPorts();
shuffleServerInfo =
new ShuffleServerInfo(
- "127.0.0.1-20001",
- shuffleServer.getIp(),
- shuffleServer.getGrpcPort(),
- shuffleServer.getNettyPort());
- }
-
- private static List<Integer> findAvailablePorts(int num) throws IOException {
- List<ServerSocket> sockets = new ArrayList<>();
- List<Integer> ports = new ArrayList<>();
-
- for (int i = 0; i < num; i++) {
- ServerSocket socket = new ServerSocket(0);
- ports.add(socket.getLocalPort());
- sockets.add(socket);
- }
-
- for (ServerSocket socket : sockets) {
- socket.close();
- }
-
- return ports;
+ LOCALHOST,
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
public void createClient(String clientType) {
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/RpcClientRetryTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/RpcClientRetryTest.java
index c2bc6234c..c1b45795c 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/RpcClientRetryTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/RpcClientRetryTest.java
@@ -44,9 +44,7 @@ import org.apache.uniffle.common.ShuffleDataDistributionType;
import org.apache.uniffle.common.ShuffleServerInfo;
import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.coordinator.CoordinatorConf;
-import org.apache.uniffle.coordinator.CoordinatorServer;
import org.apache.uniffle.server.MockedGrpcServer;
-import org.apache.uniffle.server.MockedShuffleServer;
import org.apache.uniffle.server.ShuffleServer;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.util.StorageType;
@@ -73,51 +71,30 @@ public class RpcClientRetryTest extends ShuffleReadWriteBase {
.readBufferSize(1000);
}
- public static MockedShuffleServer createMockedShuffleServer(int id, File tmpDir)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- File dataDir1 = new File(tmpDir, id + "_1");
- File dataDir2 = new File(tmpDir, id + "_2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
+ public static ShuffleServerConf buildMockShuffleConf(int id, File tmpDir) {
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(id, tmpDir, ServerType.GRPC);
shuffleServerConf.setString("rss.storage.type", StorageType.MEMORY_LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
shuffleServerConf.set(ShuffleServerConf.SERVER_MEMORY_SHUFFLE_LOWWATERMARK_PERCENTAGE, 5.0);
shuffleServerConf.set(ShuffleServerConf.SERVER_MEMORY_SHUFFLE_HIGHWATERMARK_PERCENTAGE, 15.0);
shuffleServerConf.set(ShuffleServerConf.SERVER_BUFFER_CAPACITY, 600L);
shuffleServerConf.set(ShuffleServerConf.SINGLE_BUFFER_FLUSH_BLOCKS_NUM_THRESHOLD, 1);
- return new MockedShuffleServer(shuffleServerConf);
+ return shuffleServerConf;
}
@BeforeAll
public static void initCluster(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- grpcShuffleServers.add(createMockedShuffleServer(0, tmpDir));
- grpcShuffleServers.add(createMockedShuffleServer(1, tmpDir));
- grpcShuffleServers.add(createMockedShuffleServer(2, tmpDir));
+ storeMockShuffleServerConf(buildMockShuffleConf(0, tmpDir));
+ storeMockShuffleServerConf(buildMockShuffleConf(1, tmpDir));
+ storeMockShuffleServerConf(buildMockShuffleConf(2, tmpDir));
- shuffleServerInfo0 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(0).getGrpcPort()),
- grpcShuffleServers.get(0).getIp(),
- grpcShuffleServers.get(0).getGrpcPort());
- shuffleServerInfo1 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(1).getGrpcPort()),
- grpcShuffleServers.get(1).getIp(),
- grpcShuffleServers.get(1).getGrpcPort());
- shuffleServerInfo2 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(2).getGrpcPort()),
- grpcShuffleServers.get(2).getIp(),
- grpcShuffleServers.get(2).getGrpcPort());
- for (CoordinatorServer coordinator : coordinators) {
- coordinator.start();
- }
- for (ShuffleServer shuffleServer : grpcShuffleServers) {
- shuffleServer.start();
- }
+ startServersWithRandomPorts();
+
+ shuffleServerInfo0 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
+ shuffleServerInfo1 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(1).getGrpcPort());
+ shuffleServerInfo2 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(2).getGrpcPort());
}
@AfterAll
@@ -205,6 +182,8 @@ public class RpcClientRetryTest extends ShuffleReadWriteBase {
enableFirstNReadRequestsToFail(1);
validateResult(readClient2, expectedData);
disableFirstNReadRequestsToFail();
+ readClient1.close();
+ readClient2.close();
}
private static void enableFirstNReadRequestsToFail(int failedCount) {
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleReadWriteBase.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleReadWriteBase.java
index b53df0da1..8eeca4ebe 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleReadWriteBase.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleReadWriteBase.java
@@ -17,7 +17,6 @@
package org.apache.uniffle.test;
-import java.io.File;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
@@ -133,12 +132,6 @@ public abstract class ShuffleReadWriteBase extends IntegrationTestBase {
TestUtils.validateResult(readClient, expectedData);
}
- public static String generateBasePath(File tmpDir) {
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- return dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- }
-
public static ShuffleDataResult readShuffleData(
ShuffleServerGrpcClient shuffleServerClient,
String appId,
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerConcurrentWriteOfHadoopTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerConcurrentWriteOfHadoopTest.java
index d4c41abff..7959688a3 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerConcurrentWriteOfHadoopTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerConcurrentWriteOfHadoopTest.java
@@ -17,6 +17,7 @@
package org.apache.uniffle.test;
+import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -33,6 +34,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -66,24 +68,20 @@ public class ShuffleServerConcurrentWriteOfHadoopTest extends ShuffleServerWithH
private static final int MAX_CONCURRENCY = 3;
@BeforeAll
- public static void setupServers() throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ public static void setupServers(@TempDir File tmpDir) throws Exception {
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC);
- createShuffleServer(grpcShuffleServerConf);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
- ShuffleServerConf nettyShuffleServerConf = buildShuffleServerConf(ServerType.GRPC_NETTY);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType) throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ protected static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setString(ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.HDFS.name());
shuffleServerConf.setInteger(
ShuffleServerConf.SERVER_MAX_CONCURRENCY_OF_ONE_PARTITION, MAX_CONCURRENCY);
@@ -163,10 +161,9 @@ public class ShuffleServerConcurrentWriteOfHadoopTest extends ShuffleServerWithH
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
Roaring64NavigableMap blocksBitmap = Roaring64NavigableMap.bitmapOf();
bitmaps.stream()
.forEach(
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerFaultToleranceTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerFaultToleranceTest.java
index 03c40ea69..596dd8dd6 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerFaultToleranceTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerFaultToleranceTest.java
@@ -283,8 +283,7 @@ public class ShuffleServerFaultToleranceTest extends ShuffleReadWriteBase {
return new RssSendShuffleDataRequest(appId, 3, 1000, shuffleToBlocks);
}
- public static void prepareShuffleServerConf(int subDirIndex, File tmpDir, ServerType serverType)
- throws Exception {
+ public static void prepareShuffleServerConf(int subDirIndex, File tmpDir, ServerType serverType) {
ShuffleServerConf shuffleServerConf =
shuffleServerConfWithoutPort(subDirIndex, tmpDir, serverType);
shuffleServerConf.setString(
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerGrpcTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerGrpcTest.java
index e188e04b6..fbeb6ca77 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerGrpcTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerGrpcTest.java
@@ -19,7 +19,6 @@ package org.apache.uniffle.test;
import java.io.File;
import java.io.IOException;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -92,47 +91,32 @@ public class ShuffleServerGrpcTest extends IntegrationTestBase {
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
private final AtomicInteger atomicInteger = new AtomicInteger(0);
private static final BlockIdLayout layout = BlockIdLayout.DEFAULT;
private static final Long EVENT_THRESHOLD_SIZE = 2048L;
private static final int GB = 1024 * 1024 * 1024;
protected static final long FAILED_REQUIRE_ID = -1;
- private static int rpcPort1;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
- createCoordinatorServer(coordinatorConf);
+ storeCoordinatorConf(coordinatorConf);
- File dataDir1 = new File(tmpDir, "data1");
- String grpcBasePath = dataDir1.getAbsolutePath();
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, grpcBasePath);
- rpcPort1 = grpcShuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(grpcShuffleServerConf);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
- File dataDir2 = new File(tmpDir, "data2");
- String nettyBasePath = dataDir2.getAbsolutePath();
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, nettyBasePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tmpDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tmpDir, serverType);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.MEMORY_LOCALFILE_HDFS.name());
shuffleServerConf.set(
ShuffleServerConf.FLUSH_COLD_STORAGE_THRESHOLD_SIZE, EVENT_THRESHOLD_SIZE);
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 2000L);
shuffleServerConf.set(ShuffleServerConf.SERVER_PRE_ALLOCATION_EXPIRED, 5000L);
@@ -144,13 +128,12 @@ public class ShuffleServerGrpcTest extends IntegrationTestBase {
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
@@ -178,9 +161,9 @@ public class ShuffleServerGrpcTest extends IntegrationTestBase {
.unregisterThreadPoolSize(10)
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10));
- shuffleWriteClient.registerCoordinators("127.0.0.1:" + COORDINATOR_PORT_1);
+ shuffleWriteClient.registerCoordinators(getQuorum());
shuffleWriteClient.registerShuffle(
- new ShuffleServerInfo("127.0.0.1-" + rpcPort1, "127.0.0.1", rpcPort1),
+ new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort()),
"application_clearResourceTest1",
0,
Lists.newArrayList(new PartitionRange(0, 1)),
@@ -543,8 +526,8 @@ public class ShuffleServerGrpcTest extends IntegrationTestBase {
// and ShuffleServerMetrics.TOTAL_REQUIRE_BUFFER_FAILED_FOR_HUGE_PARTITION metric should be 1
int jettyPort =
isNettyMode
- ? nettyShuffleServerConfig.getInteger(ShuffleServerConf.JETTY_HTTP_PORT)
- : grpcShuffleServerConfig.getInteger(ShuffleServerConf.JETTY_HTTP_PORT);
+ ? nettyShuffleServers.get(0).getJettyPort()
+ : grpcShuffleServers.get(0).getJettyPort();
String content =
TestUtils.httpGet(String.format("http://127.0.0.1:%s/metrics/server", jettyPort));
ObjectMapper mapper = new ObjectMapper();
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerInternalGrpcTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerInternalGrpcTest.java
index a149f4af2..f34687513 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerInternalGrpcTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerInternalGrpcTest.java
@@ -18,7 +18,6 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Lists;
@@ -54,28 +53,26 @@ public class ShuffleServerInternalGrpcTest extends IntegrationTestBase {
private ShuffleServerGrpcClient shuffleServerClient;
private ShuffleServerInternalGrpcClient shuffleServerInternalClient;
- private static int rpcPort1;
-
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- File dataDir1 = new File(tmpDir, "data1");
- String basePath = dataDir1.getAbsolutePath();
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
+ storeCoordinatorConf(coordinatorConf);
+
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC);
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 5000L);
shuffleServerConf.set(ShuffleServerConf.SERVER_DECOMMISSION_CHECK_INTERVAL, 500L);
- rpcPort1 = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
- startServers();
+ storeShuffleServerConf(shuffleServerConf);
+
+ startServersWithRandomPorts();
}
@BeforeEach
public void createClient() {
- shuffleServerClient = new ShuffleServerGrpcClient(LOCALHOST, rpcPort1);
- shuffleServerInternalClient = new ShuffleServerInternalGrpcClient(LOCALHOST, rpcPort1);
+ shuffleServerClient =
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
+ shuffleServerInternalClient =
+ new ShuffleServerInternalGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
}
@AfterEach
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerOnRandomPortTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerOnRandomPortTest.java
index eb5cefb94..c0ac00a6e 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerOnRandomPortTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerOnRandomPortTest.java
@@ -17,8 +17,11 @@
package org.apache.uniffle.test;
+import java.io.File;
+
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.apache.uniffle.common.config.RssBaseConf;
import org.apache.uniffle.common.rpc.ServerType;
@@ -30,31 +33,26 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class ShuffleServerOnRandomPortTest extends CoordinatorTestBase {
@BeforeAll
- public static void setupServers() throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
+ public static void setupServers(@TempDir File tmpDir) throws Exception {
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
coordinatorConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
coordinatorConf.setString(CoordinatorConf.COORDINATOR_ASSIGNMENT_STRATEGY.key(), "BASIC");
coordinatorConf.setLong("rss.coordinator.app.expired", 2000);
coordinatorConf.setLong("rss.coordinator.server.heartbeat.timeout", 3000);
- createCoordinatorServer(coordinatorConf);
-
- ShuffleServerConf grpcShuffleServerConf1 = buildShuffleServerConf(ServerType.GRPC);
- createShuffleServer(grpcShuffleServerConf1);
-
- ShuffleServerConf grpcShuffleServerConf2 = buildShuffleServerConf(ServerType.GRPC);
- createShuffleServer(grpcShuffleServerConf2);
-
- ShuffleServerConf nettyShuffleServerConf1 = buildShuffleServerConf(ServerType.GRPC_NETTY);
- createShuffleServer(nettyShuffleServerConf1);
+ storeCoordinatorConf(coordinatorConf);
- ShuffleServerConf nettyShuffleServerConf2 = buildShuffleServerConf(ServerType.GRPC_NETTY);
- createShuffleServer(nettyShuffleServerConf2);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(2, tmpDir, ServerType.GRPC_NETTY));
+ storeShuffleServerConf(buildShuffleServerConf(3, tmpDir, ServerType.GRPC_NETTY));
- startServers();
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType) throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setInteger("rss.server.netty.port", 0);
shuffleServerConf.setInteger("rss.rpc.server.port", 0);
shuffleServerConf.setInteger("rss.random.port.min", 30000);
@@ -63,8 +61,8 @@ public class ShuffleServerOnRandomPortTest extends CoordinatorTestBase {
}
@Test
- public void startStreamServerOnRandomPort() throws Exception {
- CoordinatorTestUtils.waitForRegister(coordinatorClient, 2);
+ public void startStreamServerOnRandomPort(@TempDir File tmpDir) throws Exception {
+ CoordinatorTestUtils.waitForRegister(coordinatorClient, 4);
Thread.sleep(5000);
int actualPort = nettyShuffleServers.get(0).getNettyPort();
assertTrue(actualPort >= 30000 && actualPort <= 40000);
@@ -72,10 +70,12 @@ public class ShuffleServerOnRandomPortTest extends CoordinatorTestBase {
assertTrue(actualPort >= 30000 && actualPort <= 40000);
int maxRetries = 100;
- ShuffleServerConf shuffleServerConf = buildShuffleServerConf(ServerType.GRPC_NETTY);
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC_NETTY);
// start netty server with already bind port
shuffleServerConf.setInteger("rss.server.netty.port", actualPort);
shuffleServerConf.setInteger("rss.port.max.retry", maxRetries);
+ shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
ShuffleServer ss = new ShuffleServer(shuffleServerConf);
ss.start();
assertTrue(ss.getNettyPort() > actualPort && actualPort <= actualPort + maxRetries);
@@ -83,8 +83,8 @@ public class ShuffleServerOnRandomPortTest extends CoordinatorTestBase {
}
@Test
- public void startGrpcServerOnRandomPort() throws Exception {
- CoordinatorTestUtils.waitForRegister(coordinatorClient, 2);
+ public void startGrpcServerOnRandomPort(@TempDir File tmpDir) throws Exception {
+ CoordinatorTestUtils.waitForRegister(coordinatorClient, 4);
Thread.sleep(5000);
int actualPort = grpcShuffleServers.get(0).getGrpcPort();
assertTrue(actualPort >= 30000 && actualPort <= 40000);
@@ -92,10 +92,12 @@ public class ShuffleServerOnRandomPortTest extends CoordinatorTestBase {
assertTrue(actualPort >= 30000 && actualPort <= 40000);
int maxRetries = 100;
- ShuffleServerConf shuffleServerConf = buildShuffleServerConf(ServerType.GRPC);
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC_NETTY);
// start grpc server with already bind port
shuffleServerConf.setInteger("rss.rpc.server.port", actualPort);
shuffleServerConf.setInteger("rss.port.max.retry", maxRetries);
+ shuffleServerConf.setString("rss.coordinator.quorum", getQuorum());
ShuffleServer ss = new ShuffleServer(shuffleServerConf);
ss.start();
assertTrue(ss.getGrpcPort() > actualPort && actualPort <= actualPort + maxRetries);
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithHadoopTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithHadoopTest.java
index 67a9dc05c..f8c05bb47 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithHadoopTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithHadoopTest.java
@@ -17,6 +17,7 @@
package org.apache.uniffle.test;
+import java.io.File;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -27,6 +28,7 @@ import com.google.common.collect.Maps;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
@@ -62,40 +64,35 @@ public class ShuffleServerWithHadoopTest extends ShuffleReadWriteBase {
protected ShuffleServerGrpcClient grpcShuffleServerClient;
protected ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- protected static ShuffleServerConf grpcShuffleServerConfig;
- protected static ShuffleServerConf nettyShuffleServerConfig;
@BeforeAll
- public static void setupServers() throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ public static void setupServers(@TempDir File tmpDir) throws Exception {
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- ShuffleServerConf grpcShuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- grpcShuffleServerConf.setString(
- ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.HDFS.name());
- createShuffleServer(grpcShuffleServerConf);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
- ShuffleServerConf nettyShuffleServerConf = getShuffleServerConf(ServerType.GRPC_NETTY);
- nettyShuffleServerConf.setString(
- ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.HDFS.name());
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
+ startServersWithRandomPorts();
+ }
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
+ shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
+ return shuffleServerConf;
}
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
@@ -162,10 +159,9 @@ public class ShuffleServerWithHadoopTest extends ShuffleReadWriteBase {
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
ShuffleReadClientImpl readClient =
baseReadBuilder(isNettyMode)
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithKerberizedHadoopTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithKerberizedHadoopTest.java
index 270b3d4b2..bcbc29f55 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithKerberizedHadoopTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithKerberizedHadoopTest.java
@@ -87,12 +87,8 @@ public class ShuffleServerWithKerberizedHadoopTest extends KerberizedHadoopBase
private static CoordinatorServer coordinatorServer;
private static ShuffleServer grpcShuffleServer;
private static ShuffleServer nettyShuffleServer;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
protected static List<Integer> jettyPorts = Lists.newArrayList();
- static @TempDir File tempDir;
-
private static ShuffleServerConf getShuffleServerConf(
int id, File tmpDir, int coordinatorRpcPort, ServerType serverType) throws Exception {
ShuffleServerConf serverConf = new ShuffleServerConf();
@@ -148,9 +144,6 @@ public class ShuffleServerWithKerberizedHadoopTest extends KerberizedHadoopBase
1, tempDir, coordinatorServer.getRpcListenPort(), ServerType.GRPC_NETTY);
nettyShuffleServer = new ShuffleServer(nettyShuffleServerConf);
nettyShuffleServer.start();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
}
@AfterAll
@@ -294,11 +287,8 @@ public class ShuffleServerWithKerberizedHadoopTest extends KerberizedHadoopBase
ShuffleServerInfo ssi =
isNettyMode
? new ShuffleServerInfo(
- LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ LOCALHOST, nettyShuffleServer.getGrpcPort(), nettyShuffleServer.getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServer.getGrpcPort());
ShuffleReadClientImpl readClient =
baseReadBuilder(isNettyMode)
.appId(appId)
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfExceptionTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfExceptionTest.java
index f51bc3649..48cb91886 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfExceptionTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfExceptionTest.java
@@ -32,7 +32,6 @@ import org.apache.uniffle.common.rpc.ServerType;
import org.apache.uniffle.coordinator.CoordinatorConf;
import org.apache.uniffle.server.ShuffleServerConf;
import org.apache.uniffle.storage.handler.impl.MemoryClientReadHandler;
-import org.apache.uniffle.storage.util.StorageType;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -45,20 +44,14 @@ public class ShuffleServerWithLocalOfExceptionTest extends ShuffleReadWriteBase
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, tmpDir, ServerType.GRPC);
shuffleServerConf.setString("rss.server.app.expired.withoutHeartbeat", "5000");
- rpcPort = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
+ storeShuffleServerConf(shuffleServerConf);
- startServers();
+ startServersWithRandomPorts();
}
@BeforeEach
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfLocalOrderTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfLocalOrderTest.java
index b0ed66d7f..545cd815b 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfLocalOrderTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalOfLocalOrderTest.java
@@ -72,38 +72,23 @@ public class ShuffleServerWithLocalOfLocalOrderTest extends ShuffleReadWriteBase
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
-
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String grpcBasePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, grpcBasePath);
- createShuffleServer(grpcShuffleServerConf);
-
- File dataDir3 = new File(tmpDir, "data3");
- File dataDir4 = new File(tmpDir, "data4");
- String nettyBasePath = dataDir3.getAbsolutePath() + "," + dataDir4.getAbsolutePath();
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, nettyBasePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
+
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
+
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
shuffleServerConf.setString("rss.server.app.expired.withoutHeartbeat", "5000");
return shuffleServerConf;
}
@@ -111,13 +96,12 @@ public class ShuffleServerWithLocalOfLocalOrderTest extends ShuffleReadWriteBase
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalTest.java
index 56c3a6a09..8c4c30aa2 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithLocalTest.java
@@ -66,38 +66,22 @@ public class ShuffleServerWithLocalTest extends ShuffleReadWriteBase {
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
-
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String grpcBasePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, grpcBasePath);
- createShuffleServer(grpcShuffleServerConf);
-
- File dataDir3 = new File(tmpDir, "data3");
- File dataDir4 = new File(tmpDir, "data4");
- String nettyBasePath = dataDir3.getAbsolutePath() + "," + dataDir4.getAbsolutePath();
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, nettyBasePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
+
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
shuffleServerConf.setString("rss.server.app.expired.withoutHeartbeat", "5000");
return shuffleServerConf;
}
@@ -105,13 +89,12 @@ public class ShuffleServerWithLocalTest extends ShuffleReadWriteBase {
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemLocalHadoopTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemLocalHadoopTest.java
index f969397f8..065dc0af2 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemLocalHadoopTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemLocalHadoopTest.java
@@ -18,7 +18,6 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
@@ -74,47 +73,33 @@ public class ShuffleServerWithMemLocalHadoopTest extends ShuffleReadWriteBase {
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
private static String REMOTE_STORAGE = HDFS_URI + "rss/test_%s";
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- File dataDir = new File(tmpDir, "data");
- String grpcBasePath = dataDir.getAbsolutePath();
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, grpcBasePath);
- createShuffleServer(grpcShuffleServerConf);
+ storeShuffleServerConf(buildShuffleServerConf(0, tmpDir, ServerType.GRPC));
+ storeShuffleServerConf(buildShuffleServerConf(1, tmpDir, ServerType.GRPC_NETTY));
- File dataDir1 = new File(tmpDir, "data1");
- String nettyBasePath = dataDir1.getAbsolutePath();
- ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, nettyBasePath);
- createShuffleServer(nettyShuffleServerConf);
-
- startServers();
-
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.MEMORY_LOCALFILE_HDFS.name());
shuffleServerConf.setLong(ShuffleServerConf.FLUSH_COLD_STORAGE_THRESHOLD_SIZE, 450L);
@@ -223,10 +208,9 @@ public class ShuffleServerWithMemLocalHadoopTest extends ShuffleReadWriteBase {
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
ComposedClientReadHandler composedClientReadHandler =
new ComposedClientReadHandler(ssi, handlers);
Map<Long, byte[]> expectedData = Maps.newHashMap();
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemoryTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemoryTest.java
index a2de8f761..e86fe1c10 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemoryTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleServerWithMemoryTest.java
@@ -18,7 +18,6 @@
package org.apache.uniffle.test;
import java.io.File;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
@@ -68,34 +67,28 @@ public class ShuffleServerWithMemoryTest extends ShuffleReadWriteBase {
private ShuffleServerGrpcClient grpcShuffleServerClient;
private ShuffleServerGrpcNettyClient nettyShuffleServerClient;
- private static ShuffleServerConf grpcShuffleServerConfig;
- private static ShuffleServerConf nettyShuffleServerConfig;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
- File dataDir = new File(tmpDir, "data");
- String basePath = dataDir.getAbsolutePath();
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
- ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(ServerType.GRPC, basePath);
- createShuffleServer(grpcShuffleServerConf);
+ ShuffleServerConf grpcShuffleServerConf = buildShuffleServerConf(0, tmpDir, ServerType.GRPC);
+ storeShuffleServerConf(grpcShuffleServerConf);
ShuffleServerConf nettyShuffleServerConf =
- buildShuffleServerConf(ServerType.GRPC_NETTY, basePath);
- createShuffleServer(nettyShuffleServerConf);
+ buildShuffleServerConf(0, tmpDir, ServerType.GRPC_NETTY);
+ storeShuffleServerConf(nettyShuffleServerConf);
- startServers();
- grpcShuffleServerConfig = grpcShuffleServerConf;
- nettyShuffleServerConfig = nettyShuffleServerConf;
+ startServersWithRandomPorts();
}
- private static ShuffleServerConf buildShuffleServerConf(ServerType serverType, String basePath)
- throws Exception {
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(serverType);
+ private static ShuffleServerConf buildShuffleServerConf(
+ int subDirIndex, File tempDir, ServerType serverType) {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tempDir, serverType);
shuffleServerConf.setString(
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.LOCALFILE.name());
- shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 5000L);
shuffleServerConf.set(ShuffleServerConf.SERVER_MEMORY_SHUFFLE_LOWWATERMARK_PERCENTAGE, 5.0);
shuffleServerConf.set(ShuffleServerConf.SERVER_MEMORY_SHUFFLE_HIGHWATERMARK_PERCENTAGE, 15.0);
@@ -107,13 +100,12 @@ public class ShuffleServerWithMemoryTest extends ShuffleReadWriteBase {
@BeforeEach
public void createClient() throws Exception {
grpcShuffleServerClient =
- new ShuffleServerGrpcClient(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ new ShuffleServerGrpcClient(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
nettyShuffleServerClient =
new ShuffleServerGrpcNettyClient(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort());
}
@AfterEach
@@ -221,10 +213,9 @@ public class ShuffleServerWithMemoryTest extends ShuffleReadWriteBase {
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
ComposedClientReadHandler composedClientReadHandler =
new ComposedClientReadHandler(ssi, handlers);
// read from memory with ComposedClientReadHandler
@@ -431,10 +422,9 @@ public class ShuffleServerWithMemoryTest extends ShuffleReadWriteBase {
isNettyMode
? new ShuffleServerInfo(
LOCALHOST,
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT),
- nettyShuffleServerConfig.getInteger(ShuffleServerConf.NETTY_SERVER_PORT))
- : new ShuffleServerInfo(
- LOCALHOST, grpcShuffleServerConfig.getInteger(ShuffleServerConf.RPC_SERVER_PORT));
+ nettyShuffleServers.get(0).getGrpcPort(),
+ nettyShuffleServers.get(0).getNettyPort())
+ : new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
ComposedClientReadHandler composedClientReadHandler =
new ComposedClientReadHandler(ssi, handlers);
Map<Long, byte[]> expectedData = Maps.newHashMap();
diff --git a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleWithRssClientTest.java b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleWithRssClientTest.java
index 595f5d2d3..3d8b6e5c2 100644
--- a/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleWithRssClientTest.java
+++ b/integration-test/common/src/test/java/org/apache/uniffle/test/ShuffleWithRssClientTest.java
@@ -64,41 +64,25 @@ public class ShuffleWithRssClientTest extends ShuffleReadWriteBase {
private static ShuffleServerInfo shuffleServerInfo1;
private static ShuffleServerInfo shuffleServerInfo2;
private ShuffleWriteClientImpl shuffleWriteClientImpl;
- private static int rpcPort1;
@BeforeAll
public static void setupServers(@TempDir File tmpDir) throws Exception {
- CoordinatorConf coordinatorConf = getCoordinatorConf();
- createCoordinatorServer(coordinatorConf);
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
- shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 4000);
- File dataDir1 = new File(tmpDir, "data1");
- File dataDir2 = new File(tmpDir, "data2");
- String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
- shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
- shuffleServerConf.setString("rss.storage.basePath", basePath);
- rpcPort1 = shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT);
- createShuffleServer(shuffleServerConf);
- File dataDir3 = new File(tmpDir, "data3");
- File dataDir4 = new File(tmpDir, "data4");
- basePath = dataDir3.getAbsolutePath() + "," + dataDir4.getAbsolutePath();
- shuffleServerConf.setString("rss.storage.basePath", basePath);
- shuffleServerConf.setInteger(
- "rss.rpc.server.port", shuffleServerConf.getInteger(ShuffleServerConf.RPC_SERVER_PORT) + 1);
- shuffleServerConf.setInteger(
- "rss.jetty.http.port", shuffleServerConf.getInteger(ShuffleServerConf.JETTY_HTTP_PORT) + 1);
- createShuffleServer(shuffleServerConf);
- startServers();
- shuffleServerInfo1 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(0).getGrpcPort()),
- grpcShuffleServers.get(0).getIp(),
- grpcShuffleServers.get(0).getGrpcPort());
- shuffleServerInfo2 =
- new ShuffleServerInfo(
- String.format("127.0.0.1-%s", grpcShuffleServers.get(0).getGrpcPort() + 1),
- grpcShuffleServers.get(1).getIp(),
- grpcShuffleServers.get(0).getGrpcPort() + 1);
+ CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
+ storeCoordinatorConf(coordinatorConf);
+
+ Lists.newArrayList(0, 1)
+ .forEach(
+ subDirIndex -> {
+ ShuffleServerConf shuffleServerConf =
+ shuffleServerConfWithoutPort(subDirIndex, tmpDir, ServerType.GRPC);
+ shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 4000);
+ shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
+ storeShuffleServerConf(shuffleServerConf);
+ });
+
+ startServersWithRandomPorts();
+ shuffleServerInfo1 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(0).getGrpcPort());
+ shuffleServerInfo2 = new ShuffleServerInfo(LOCALHOST, grpcShuffleServers.get(1).getGrpcPort());
}
@BeforeEach
@@ -142,7 +126,10 @@ public class ShuffleWithRssClientTest extends ShuffleReadWriteBase {
// simulator a failed server
ShuffleServerInfo fakeShuffleServerInfo =
- new ShuffleServerInfo("127.0.0.1-20001", grpcShuffleServers.get(0).getIp(), rpcPort1 + 100);
+ new ShuffleServerInfo(
+ "127.0.0.1-20001",
+ grpcShuffleServers.get(0).getIp(),
+ generateNonExistingPorts(1).get(0));
List<ShuffleBlockInfo> blocks =
createShuffleBlockList(
0,
@@ -403,7 +390,9 @@ public class ShuffleWithRssClientTest extends ShuffleReadWriteBase {
// commit will be failed because of fakeIp
commitResult =
shuffleWriteClientImpl.sendCommit(
- Sets.newHashSet(new ShuffleServerInfo("127.0.0.1-20001", "fakeIp", rpcPort1)),
+ Sets.newHashSet(
+ new ShuffleServerInfo(
+ "127.0.0.1-20001", "fakeIp", grpcShuffleServers.get(0).getGrpcPort())),
testAppId,
0,
2);
@@ -445,10 +434,10 @@ public class ShuffleWithRssClientTest extends ShuffleReadWriteBase {
String appId = "app-1";
RemoteStorageInfo remoteStorage = new RemoteStorageInfo("");
ShuffleAssignmentsInfo response = null;
- ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
+ ShuffleServerConf shuffleServerConf = shuffleServerConfWithoutPort(0, null, ServerType.GRPC);
int heartbeatInterval = shuffleServerConf.getInteger("rss.server.heartbeat.interval", 1000);
Thread.sleep(heartbeatInterval * 2);
- shuffleWriteClientImpl.registerCoordinators(COORDINATOR_QUORUM);
+ shuffleWriteClientImpl.registerCoordinators(getQuorum());
response =
RetryUtils.retry(
() -> {