Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#2385] improvement(test): Make all tests in integration-common-test use random port #2387

Open
wants to merge 7 commits into
base: master
Choose a base branch
from
+506 −844
Open

[#2385] improvement(test): Make all tests in integration-common-test use random port #2387

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e00fe08
integration-common use random port
summaryzb Mar 10, 2025
2669650
add SuppressWarnings
summaryzb Mar 10, 2025
b1709d8
fix
summaryzb Mar 10, 2025
fd38461
remove fixed port
summaryzb Mar 10, 2025
35c9288
fix
summaryzb Mar 12, 2025
5379936
fix3
summaryzb Mar 12, 2025
0801162
fix3
summaryzb Mar 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 32 additions & 73 deletions integration-test/common/src/test/java/org/apache/uniffle/test/AssignmentWithTagsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -32,8 +30,6 @@
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;
Expand All @@ -42,7 +38,6 @@
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;
Expand All @@ -56,87 +51,51 @@
* {@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);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with this pr, some test helper methods will become unused. Could you please clean them up together in current one? such as AssignmentWithTagsTest#findAvailablePorts

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) {
Expand All @@ -151,7 +110,7 @@ public static void setupServers(@TempDir File tmpDir) throws Exception {
}

@Test
public void testTags() throws Exception {
public void testTags() {
ShuffleWriteClientImpl shuffleWriteClient =
ShuffleClientFactory.newWriteBuilder()
.clientType(ClientType.GRPC.name())
Expand All @@ -168,7 +127,7 @@ public void testTags() throws Exception {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
shuffleWriteClient.registerCoordinators(COORDINATOR_QUORUM);
shuffleWriteClient.registerCoordinators(getQuorum());

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

// 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 {
Expand All @@ -227,7 +186,7 @@ public void testTags() throws Exception {
1,
1,
1,
Sets.newHashSet("fixed", "elastic", Constants.SHUFFLE_SERVER_VERSION),
Sets.newHashSet(tag1, tag2, Constants.SHUFFLE_SERVER_VERSION),
1,
-1);
fail();
Expand Down
12 changes: 4 additions & 8 deletions ...ration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAdminServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,6 @@
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";

Expand All @@ -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());
}

Expand Down
33 changes: 11 additions & 22 deletions integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorAssignmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
Expand Down Expand Up @@ -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);
}
Expand All @@ -112,7 +101,7 @@ public void testSilentPeriod() throws Exception {
.unregisterTimeSec(10)
.unregisterRequestTimeSec(10)
.build();
shuffleWriteClient.registerCoordinators(QUORUM);
shuffleWriteClient.registerCoordinators(getQuorum());

// Case1: Disable silent period
ShuffleAssignmentsInfo info =
Expand Down Expand Up @@ -154,7 +143,7 @@ public void testAssignmentServerNodesNumber() throws Exception {
.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
Expand Down Expand Up @@ -204,7 +193,7 @@ public void testGetReShuffleAssignments() {
.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());
Expand Down
10 changes: 5 additions & 5 deletions integration-test/common/src/test/java/org/apache/uniffle/test/CoordinatorGrpcServerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ public void registerApplicationInfo(
@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);
Expand All @@ -70,17 +70,17 @@ public void testGrpcConnectionSize() throws Exception {
.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();
assertEquals(0, connSize);

// 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"));

Expand Down
Loading
Loading