Skip to content

Commit 87362fb

Browse files
summaryzbsummaryzb
authored
[#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
1 parent f65d09e commit 87362fb

34 files changed

+506
-844
lines changed

‎integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java

+32-73
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,7 @@
1818
package org.apache.uniffle.test;
1919

2020
import java.io.File;
21-
import java.io.IOException;
22-
import java.net.ServerSocket;
23-
import java.util.ArrayList;
21+
import java.util.Arrays;
2422
import java.util.HashMap;
2523
import java.util.List;
2624
import java.util.Map;
@@ -32,8 +30,6 @@
3230
import org.junit.jupiter.api.BeforeAll;
3331
import org.junit.jupiter.api.Test;
3432
import org.junit.jupiter.api.io.TempDir;
35-
import org.slf4j.Logger;
36-
import org.slf4j.LoggerFactory;
3733

3834
import org.apache.uniffle.client.factory.ShuffleClientFactory;
3935
import org.apache.uniffle.client.impl.ShuffleWriteClientImpl;
@@ -42,7 +38,6 @@
4238
import org.apache.uniffle.common.rpc.ServerType;
4339
import org.apache.uniffle.common.util.Constants;
4440
import org.apache.uniffle.coordinator.CoordinatorConf;
45-
import org.apache.uniffle.coordinator.CoordinatorServer;
4641
import org.apache.uniffle.server.ShuffleServer;
4742
import org.apache.uniffle.server.ShuffleServerConf;
4843
import org.apache.uniffle.storage.util.StorageType;
@@ -56,87 +51,51 @@
5651
* {@code RssClientConfig.RSS_CLIENT_ASSIGNMENT_TAGS}
5752
*/
5853
public class AssignmentWithTagsTest extends CoordinatorTestBase {
59-
private static final Logger LOG = LoggerFactory.getLogger(AssignmentWithTagsTest.class);
6054

6155
// KV: tag -> shuffle server id
6256
private static Map<String, List<Integer>> tagOfShufflePorts = new HashMap<>();
57+
private static final String tag1 = "fixed";
58+
private static final String tag2 = "elastic";
6359

64-
private static List<Integer> findAvailablePorts(int num) throws IOException {
65-
List<ServerSocket> sockets = new ArrayList<>();
66-
List<Integer> ports = new ArrayList<>();
67-
68-
for (int i = 0; i < num; i++) {
69-
ServerSocket socket = new ServerSocket(0);
70-
ports.add(socket.getLocalPort());
71-
sockets.add(socket);
72-
}
73-
74-
for (ServerSocket socket : sockets) {
75-
socket.close();
76-
}
77-
78-
return ports;
79-
}
80-
81-
private static void createAndStartShuffleServerWithTags(Set<String> tags, File tmpDir)
82-
throws Exception {
83-
ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
60+
private static void prepareShuffleServerConf(int subDirIndex, Set<String> tags, File tmpDir) {
61+
ShuffleServerConf shuffleServerConf =
62+
shuffleServerConfWithoutPort(subDirIndex, tmpDir, ServerType.GRPC);
8463
shuffleServerConf.setLong("rss.server.app.expired.withoutHeartbeat", 4000);
85-
86-
File dataDir1 = new File(tmpDir, "data1");
87-
File dataDir2 = new File(tmpDir, "data2");
88-
String basePath = dataDir1.getAbsolutePath() + "," + dataDir2.getAbsolutePath();
89-
9064
shuffleServerConf.setString("rss.storage.type", StorageType.LOCALFILE.name());
91-
shuffleServerConf.setString("rss.storage.basePath", basePath);
9265
shuffleServerConf.setString("rss.server.tags", StringUtils.join(tags, ","));
93-
94-
List<Integer> ports = findAvailablePorts(2);
95-
shuffleServerConf.setInteger("rss.rpc.server.port", ports.get(0));
96-
shuffleServerConf.setInteger("rss.jetty.http.port", ports.get(1));
97-
98-
for (String tag : tags) {
99-
tagOfShufflePorts.putIfAbsent(tag, new ArrayList<>());
100-
tagOfShufflePorts.get(tag).add(ports.get(0));
101-
}
102-
tagOfShufflePorts.putIfAbsent(Constants.SHUFFLE_SERVER_VERSION, new ArrayList<>());
103-
tagOfShufflePorts.get(Constants.SHUFFLE_SERVER_VERSION).add(ports.get(0));
104-
105-
LOG.info(
106-
"Shuffle server data dir: {}, rpc port: {}, http port: {}",
107-
dataDir1 + "," + dataDir2,
108-
ports.get(0),
109-
ports.get(1));
110-
111-
ShuffleServer server = new ShuffleServer(shuffleServerConf);
112-
grpcShuffleServers.add(server);
113-
server.start();
66+
storeShuffleServerConf(shuffleServerConf);
11467
}
11568

11669
@BeforeAll
11770
public static void setupServers(@TempDir File tmpDir) throws Exception {
118-
CoordinatorConf coordinatorConf = getCoordinatorConf();
119-
createCoordinatorServer(coordinatorConf);
71+
CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
72+
storeCoordinatorConf(coordinatorConf);
12073

121-
for (CoordinatorServer coordinator : coordinators) {
122-
coordinator.start();
123-
}
124-
125-
File dir1 = new File(tmpDir, "server1");
12674
for (int i = 0; i < 2; i++) {
127-
createAndStartShuffleServerWithTags(Sets.newHashSet(), dir1);
75+
prepareShuffleServerConf(i, Sets.newHashSet(), tmpDir);
12876
}
12977

130-
File dir2 = new File(tmpDir, "server2");
13178
for (int i = 0; i < 2; i++) {
132-
createAndStartShuffleServerWithTags(Sets.newHashSet("fixed"), dir2);
79+
prepareShuffleServerConf(2 + i, Sets.newHashSet(tag1), tmpDir);
13380
}
13481

135-
File dir3 = new File(tmpDir, "server3");
13682
for (int i = 0; i < 2; i++) {
137-
createAndStartShuffleServerWithTags(Sets.newHashSet("elastic"), dir3);
83+
prepareShuffleServerConf(4 + i, Sets.newHashSet(tag2), tmpDir);
13884
}
13985

86+
startServersWithRandomPorts();
87+
List<Integer> collect =
88+
grpcShuffleServers.stream().map(ShuffleServer::getGrpcPort).collect(Collectors.toList());
89+
tagOfShufflePorts.put(Constants.SHUFFLE_SERVER_VERSION, collect);
90+
tagOfShufflePorts.put(
91+
tag1,
92+
Arrays.asList(
93+
grpcShuffleServers.get(2).getGrpcPort(), grpcShuffleServers.get(3).getGrpcPort()));
94+
tagOfShufflePorts.put(
95+
tag2,
96+
Arrays.asList(
97+
grpcShuffleServers.get(4).getGrpcPort(), grpcShuffleServers.get(5).getGrpcPort()));
98+
14099
// Wait all shuffle servers registering to coordinator
141100
long startTimeMS = System.currentTimeMillis();
142101
while (true) {
@@ -151,7 +110,7 @@ public static void setupServers(@TempDir File tmpDir) throws Exception {
151110
}
152111

153112
@Test
154-
public void testTags() throws Exception {
113+
public void testTags() {
155114
ShuffleWriteClientImpl shuffleWriteClient =
156115
ShuffleClientFactory.newWriteBuilder()
157116
.clientType(ClientType.GRPC.name())
@@ -168,7 +127,7 @@ public void testTags() throws Exception {
168127
.unregisterTimeSec(10)
169128
.unregisterRequestTimeSec(10)
170129
.build();
171-
shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
130+
shuffleWriteClient.registerCoordinators(getQuorum());
172131

173132
// Case1 : only set the single default shuffle version tag
174133
ShuffleAssignmentsInfo assignmentsInfo =
@@ -198,26 +157,26 @@ public void testTags() throws Exception {
198157

199158
// Case3: Set the single fixed tag
200159
assignmentsInfo =
201-
shuffleWriteClient.getShuffleAssignments("app-3", 1, 1, 1, Sets.newHashSet("fixed"), 1, -1);
160+
shuffleWriteClient.getShuffleAssignments("app-3", 1, 1, 1, Sets.newHashSet(tag1), 1, -1);
202161
assignedServerPorts =
203162
assignmentsInfo.getPartitionToServers().values().stream()
204163
.flatMap(x -> x.stream())
205164
.map(x -> x.getGrpcPort())
206165
.collect(Collectors.toList());
207166
assertEquals(1, assignedServerPorts.size());
208-
assertTrue(tagOfShufflePorts.get("fixed").contains(assignedServerPorts.get(0)));
167+
assertTrue(tagOfShufflePorts.get(tag1).contains(assignedServerPorts.get(0)));
209168

210169
// case4: Set the multiple tags if exists
211170
assignmentsInfo =
212171
shuffleWriteClient.getShuffleAssignments(
213-
"app-4", 1, 1, 1, Sets.newHashSet("fixed", Constants.SHUFFLE_SERVER_VERSION), 1, -1);
172+
"app-4", 1, 1, 1, Sets.newHashSet(tag1, Constants.SHUFFLE_SERVER_VERSION), 1, -1);
214173
assignedServerPorts =
215174
assignmentsInfo.getPartitionToServers().values().stream()
216175
.flatMap(x -> x.stream())
217176
.map(x -> x.getGrpcPort())
218177
.collect(Collectors.toList());
219178
assertEquals(1, assignedServerPorts.size());
220-
assertTrue(tagOfShufflePorts.get("fixed").contains(assignedServerPorts.get(0)));
179+
assertTrue(tagOfShufflePorts.get(tag1).contains(assignedServerPorts.get(0)));
221180

222181
// case5: Set the multiple tags if non-exist
223182
try {
@@ -227,7 +186,7 @@ public void testTags() throws Exception {
227186
1,
228187
1,
229188
1,
230-
Sets.newHashSet("fixed", "elastic", Constants.SHUFFLE_SERVER_VERSION),
189+
Sets.newHashSet(tag1, tag2, Constants.SHUFFLE_SERVER_VERSION),
231190
1,
232191
-1);
233192
fail();

‎integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java

+4-8
Original file line numberDiff line numberDiff line change
@@ -34,8 +34,6 @@
3434
import static org.junit.jupiter.api.Assertions.assertEquals;
3535

3636
public class CoordinatorAdminServiceTest extends IntegrationTestBase {
37-
38-
private static final Integer JETTY_HTTP_PORT = 12345;
3937
private static final String accessChecker =
4038
"org.apache.uniffle.test.AccessClusterTest$MockedAccessChecker";
4139

@@ -45,18 +43,16 @@ public class CoordinatorAdminServiceTest extends IntegrationTestBase {
4543

4644
@BeforeAll
4745
public static void setUp() throws Exception {
48-
CoordinatorConf coordinatorConf = new CoordinatorConf();
49-
coordinatorConf.set(RssBaseConf.JETTY_HTTP_PORT, JETTY_HTTP_PORT);
46+
CoordinatorConf coordinatorConf = coordinatorConfWithoutPort();
5047
coordinatorConf.set(RssBaseConf.JETTY_CORE_POOL_SIZE, 128);
51-
coordinatorConf.set(RssBaseConf.RPC_SERVER_PORT, 12346);
5248
coordinatorConf.setString(CoordinatorConf.COORDINATOR_ACCESS_CHECKERS.key(), accessChecker);
53-
createCoordinatorServer(coordinatorConf);
54-
startServers();
49+
storeCoordinatorConf(coordinatorConf);
50+
startServersWithRandomPorts();
5551
}
5652

5753
@BeforeEach
5854
public void createClient() {
59-
String hostUrl = String.format("http://%s:%d", LOCALHOST, JETTY_HTTP_PORT);
55+
String hostUrl = String.format("http://%s:%d", LOCALHOST, jettyPorts.get(0));
6056
adminRestApi = new AdminRestApi(UniffleRestClient.builder(hostUrl).build());
6157
}
6258

‎integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java

+11-22
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
import java.io.File;
2121
import java.util.ArrayList;
22-
import java.util.Arrays;
2322
import java.util.HashSet;
2423
import java.util.List;
2524
import java.util.Set;
@@ -54,42 +53,32 @@ public class CoordinatorAssignmentTest extends CoordinatorTestBase {
5453
private static final int SERVER_NUM = 10;
5554
private static final HashSet<String> TAGS = Sets.newHashSet("t1");
5655

57-
private static final String QUORUM =
58-
LOCALHOST + ":" + COORDINATOR_PORT_1 + "," + LOCALHOST + ":" + COORDINATOR_PORT_2;
59-
6056
@BeforeAll
6157
public static void setupServers(@TempDir File tmpDir) throws Exception {
62-
CoordinatorConf coordinatorConf1 = getCoordinatorConf();
58+
CoordinatorConf coordinatorConf1 = coordinatorConfWithoutPort();
6359
coordinatorConf1.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
6460
coordinatorConf1.setInteger(CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, SHUFFLE_NODES_MAX);
6561
coordinatorConf1.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
66-
createCoordinatorServer(coordinatorConf1);
62+
storeCoordinatorConf(coordinatorConf1);
6763

68-
CoordinatorConf coordinatorConf2 = getCoordinatorConf();
64+
CoordinatorConf coordinatorConf2 = coordinatorConfWithoutPort();
6965
coordinatorConf2.setLong(CoordinatorConf.COORDINATOR_APP_EXPIRED, 2000);
7066
coordinatorConf2.setInteger(CoordinatorConf.COORDINATOR_SHUFFLE_NODES_MAX, SHUFFLE_NODES_MAX);
71-
coordinatorConf2.setInteger(CoordinatorConf.RPC_SERVER_PORT, COORDINATOR_PORT_2);
72-
coordinatorConf2.setInteger(CoordinatorConf.JETTY_HTTP_PORT, JETTY_PORT_2);
7367
coordinatorConf2.setLong(RssBaseConf.RSS_RECONFIGURE_INTERVAL_SEC, 1L);
74-
createCoordinatorServer(coordinatorConf2);
68+
storeCoordinatorConf(coordinatorConf2);
7569

7670
for (int i = 0; i < SERVER_NUM; i++) {
77-
ShuffleServerConf shuffleServerConf = getShuffleServerConf(ServerType.GRPC);
78-
File dataDir = new File(tmpDir, "data" + i);
79-
String basePath = dataDir.getAbsolutePath();
71+
ShuffleServerConf shuffleServerConf =
72+
shuffleServerConfWithoutPort(i, tmpDir, ServerType.GRPC);
8073
shuffleServerConf.setString(
8174
ShuffleServerConf.RSS_STORAGE_TYPE.key(), StorageType.MEMORY_LOCALFILE_HDFS.name());
82-
shuffleServerConf.set(ShuffleServerConf.RSS_STORAGE_BASE_PATH, Arrays.asList(basePath));
8375
shuffleServerConf.set(RssBaseConf.RPC_METRICS_ENABLED, true);
8476
shuffleServerConf.set(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT, 2000L);
8577
shuffleServerConf.set(ShuffleServerConf.SERVER_PRE_ALLOCATION_EXPIRED, 5000L);
86-
shuffleServerConf.setInteger(RssBaseConf.RPC_SERVER_PORT, 18001 + i);
87-
shuffleServerConf.setInteger(RssBaseConf.JETTY_HTTP_PORT, 19010 + i);
8878
shuffleServerConf.set(ShuffleServerConf.TAGS, new ArrayList<>(TAGS));
89-
shuffleServerConf.setString("rss.coordinator.quorum", QUORUM);
90-
createShuffleServer(shuffleServerConf);
79+
storeShuffleServerConf(shuffleServerConf);
9180
}
92-
startServers();
81+
startServersWithRandomPorts();
9382

9483
Thread.sleep(1000 * 5);
9584
}
@@ -112,7 +101,7 @@ public void testSilentPeriod() throws Exception {
112101
.unregisterTimeSec(10)
113102
.unregisterRequestTimeSec(10)
114103
.build();
115-
shuffleWriteClient.registerCoordinators(QUORUM);
104+
shuffleWriteClient.registerCoordinators(getQuorum());
116105

117106
// Case1: Disable silent period
118107
ShuffleAssignmentsInfo info =
@@ -154,7 +143,7 @@ public void testAssignmentServerNodesNumber() throws Exception {
154143
.unregisterTimeSec(10)
155144
.unregisterRequestTimeSec(10)
156145
.build();
157-
shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
146+
shuffleWriteClient.registerCoordinators(getQuorum());
158147

159148
/**
160149
* case1: user specify the illegal shuffle node num(<0) it will use the default shuffle nodes
@@ -204,7 +193,7 @@ public void testGetReShuffleAssignments() {
204193
.unregisterTimeSec(10)
205194
.unregisterRequestTimeSec(10)
206195
.build();
207-
shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
196+
shuffleWriteClient.registerCoordinators(getQuorum());
208197
Set<String> excludeServer = Sets.newConcurrentHashSet();
209198
List<ShuffleServer> excludeShuffleServer =
210199
grpcShuffleServers.stream().limit(3).collect(Collectors.toList());

‎integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ public void registerApplicationInfo(
5858
@Test
5959
public void testGrpcConnectionSize() throws Exception {
6060
RssBaseConf baseConf = new RssBaseConf();
61-
baseConf.set(RssBaseConf.RPC_SERVER_PORT, 20001);
61+
baseConf.set(RssBaseConf.RPC_SERVER_PORT, 0);
6262
baseConf.set(RssBaseConf.RPC_EXECUTOR_SIZE, 2);
6363

6464
GRPCMetrics grpcMetrics = new CoordinatorGrpcMetrics(baseConf);
@@ -70,17 +70,17 @@ public void testGrpcConnectionSize() throws Exception {
7070
.addService(new MockedCoordinatorGrpcService())
7171
.build();
7272
try {
73-
grpcServer.start();
73+
final int port = grpcServer.start();
7474

7575
// case1: test the single one connection metric
7676
double connSize = grpcMetrics.getGaugeMap().get(GRPC_SERVER_CONNECTION_NUMBER_KEY).get();
7777
assertEquals(0, connSize);
7878

7979
// case2: test the multiple connections
8080
try (CoordinatorGrpcClient coordinatorGrpcClient =
81-
new CoordinatorGrpcClient("localhost", 20001);
82-
CoordinatorGrpcClient client1 = new CoordinatorGrpcClient("localhost", 20001);
83-
CoordinatorGrpcClient client2 = new CoordinatorGrpcClient("localhost", 20001)) {
81+
new CoordinatorGrpcClient("localhost", port);
82+
CoordinatorGrpcClient client1 = new CoordinatorGrpcClient("localhost", port);
83+
CoordinatorGrpcClient client2 = new CoordinatorGrpcClient("localhost", port)) {
8484
coordinatorGrpcClient.registerApplicationInfo(
8585
new RssApplicationInfoRequest("testGrpcConnectionSize", 10000, "user"));
8686

0 commit comments

Comments
 (0)