You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2024/01/27 04:52:11 UTC
(incubator-celeborn) branch main updated: [CELEBORN-1235] Start test nodes in random ports to allow multiple builds run in the same ci server
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 916df8f6e [CELEBORN-1235] Start test nodes in random ports to allow multiple builds run in the same ci server
916df8f6e is described below
commit 916df8f6ecf06d6b394dd6e3c0b06356eb5d694a
Author: CodingCat <zh...@gmail.com>
AuthorDate: Sat Jan 27 12:51:53 2024 +0800
[CELEBORN-1235] Start test nodes in random ports to allow multiple builds run in the same ci server
### What changes were proposed in this pull request?
this PR is to improve the test implementations so that it starts test nodes in random ports instead of using the hardcoded ones
### Why are the changes needed?
currently the test nodes are started in the hard coded ports, this prevents to run multiple builds in the same CI/CD server (which is not uncommonly seen in many companies infra)
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
it runs in our private CI/CD infra with many parallel builds very well
Closes #2237 from CodingCat/enhance_test.
Authored-by: CodingCat <zh...@gmail.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../plugin/flink/RemoteShuffleMasterTest.java | 4 +
.../plugin/flink/RemoteShuffleMasterTest.java | 4 +
.../plugin/flink/RemoteShuffleMasterTest.java | 4 +
.../plugin/flink/RemoteShuffleMasterTest.java | 4 +
.../celeborn/common/client/MasterClient.java | 5 +-
.../org/apache/celeborn/common/util/Utils.scala | 6 +-
.../ha/RatisMasterStatusSystemSuiteJ.java | 178 ++++++++++++---------
.../service/deploy/master/MasterSuite.scala | 8 +-
.../apache/celeborn/tests/flink/SplitTest.scala | 14 +-
.../celeborn/tests/flink/WordCountTest.scala | 12 +-
.../apache/celeborn/tests/mr/WordCountTest.scala | 117 ++++++++------
.../client/LifecycleManagerCommitFilesSuite.scala | 13 +-
.../client/LifecycleManagerDestroySlotsSuite.scala | 13 +-
.../LifecycleManagerSetupEndpointSuite.scala | 12 +-
.../tests/client/LifecycleManagerSuite.scala | 16 +-
.../celeborn/tests/client/ShuffleClientSuite.scala | 27 ++--
.../tests/spark/CelebornFetchFailureSuite.scala | 17 +-
.../tests/spark/CelebornHashCheckDiskSuite.scala | 2 +-
.../celeborn/tests/spark/PushDataTimeoutTest.scala | 2 +-
.../celeborn/tests/spark/RetryReviveTest.scala | 2 +-
.../tests/spark/ShuffleFallbackSuite.scala | 4 +-
.../spark/SlotsAssignMaxWorkersLargeTest.scala | 5 +-
.../spark/SlotsAssignMaxWorkersSmallTest.scala | 5 +-
.../celeborn/tests/spark/SparkTestBase.scala | 2 +-
.../deploy/worker/storage/StorageManager.scala | 1 +
.../celeborn/service/deploy/HeartbeatFeature.scala | 9 +-
.../service/deploy/MiniClusterFeature.scala | 148 +++++++++++++++--
.../service/deploy/cluster/ReadWriteTestBase.scala | 11 +-
28 files changed, 408 insertions(+), 237 deletions(-)
diff --git a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 7cb6139f1..6d96a3b12 100644
--- a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++ b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -46,6 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.Utils$;
import org.apache.celeborn.plugin.flink.utils.FlinkUtils;
public class RemoteShuffleMasterTest {
@@ -57,6 +58,9 @@ public class RemoteShuffleMasterTest {
@Before
public void setUp() {
configuration = new Configuration();
+ int startPort = Utils$.MODULE$.selectRandomPort(1024, 65535);
+ configuration.setInteger("celeborn.master.port", startPort);
+ configuration.setString("celeborn.master.endpoints", "localhost:" + startPort);
remoteShuffleMaster = createShuffleMaster(configuration);
}
diff --git a/client-flink/flink-1.15/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java b/client-flink/flink-1.15/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 7cb6139f1..6d96a3b12 100644
--- a/client-flink/flink-1.15/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++ b/client-flink/flink-1.15/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -46,6 +46,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.Utils$;
import org.apache.celeborn.plugin.flink.utils.FlinkUtils;
public class RemoteShuffleMasterTest {
@@ -57,6 +58,9 @@ public class RemoteShuffleMasterTest {
@Before
public void setUp() {
configuration = new Configuration();
+ int startPort = Utils$.MODULE$.selectRandomPort(1024, 65535);
+ configuration.setInteger("celeborn.master.port", startPort);
+ configuration.setString("celeborn.master.endpoints", "localhost:" + startPort);
remoteShuffleMaster = createShuffleMaster(configuration);
}
diff --git a/client-flink/flink-1.17/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java b/client-flink/flink-1.17/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 8cdf82c79..18a867dbb 100644
--- a/client-flink/flink-1.17/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++ b/client-flink/flink-1.17/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -53,6 +53,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.Utils$;
import org.apache.celeborn.plugin.flink.utils.FlinkUtils;
public class RemoteShuffleMasterTest {
@@ -64,6 +65,9 @@ public class RemoteShuffleMasterTest {
@Before
public void setUp() {
configuration = new Configuration();
+ int startPort = Utils$.MODULE$.selectRandomPort(1024, 65535);
+ configuration.setInteger("celeborn.master.port", startPort);
+ configuration.setString("celeborn.master.endpoints", "localhost:" + startPort);
remoteShuffleMaster = createShuffleMaster(configuration);
}
diff --git a/client-flink/flink-1.18/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java b/client-flink/flink-1.18/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 8cdf82c79..18a867dbb 100644
--- a/client-flink/flink-1.18/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++ b/client-flink/flink-1.18/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -53,6 +53,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.util.Utils$;
import org.apache.celeborn.plugin.flink.utils.FlinkUtils;
public class RemoteShuffleMasterTest {
@@ -64,6 +65,9 @@ public class RemoteShuffleMasterTest {
@Before
public void setUp() {
configuration = new Configuration();
+ int startPort = Utils$.MODULE$.selectRandomPort(1024, 65535);
+ configuration.setInteger("celeborn.master.port", startPort);
+ configuration.setString("celeborn.master.endpoints", "localhost:" + startPort);
remoteShuffleMaster = createShuffleMaster(configuration);
}
diff --git a/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java b/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
index 2e8468d23..3c0c90f07 100644
--- a/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
+++ b/common/src/main/java/org/apache/celeborn/common/client/MasterClient.java
@@ -237,8 +237,9 @@ public class MasterClient {
if (endpointRef == null) {
throw new IllegalStateException(
- "After trying all the available Master Addresses,"
- + " an usable link still couldn't be created.");
+ "After trying all the available Master Addresses("
+ + String.join(",", masterEndpoints)
+ + "), an usable link still couldn't be created.");
} else {
LOG.info("connect to master {}.", endpointRef.address());
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index 125e1f67e..2b71e8cb8 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -32,7 +32,7 @@ import scala.annotation.tailrec
import scala.collection.JavaConverters._
import scala.io.Source
import scala.reflect.ClassTag
-import scala.util.Try
+import scala.util.{Random => ScalaRandom, Try}
import scala.util.control.{ControlThrowable, NonFatal}
import com.google.protobuf.{ByteString, GeneratedMessageV3}
@@ -243,6 +243,10 @@ object Utils extends Logging {
}
}
+ def selectRandomPort(from: Int, to: Int): Int = {
+ ScalaRandom.nextInt(to - from) + from
+ }
+
def startServiceOnPort[T](
startPort: Int,
startService: Int => (T, Int),
diff --git a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
index b66099e1d..c4fb8e29a 100644
--- a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
+++ b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
@@ -40,6 +40,7 @@ import org.apache.celeborn.common.rpc.RpcEndpointRef;
import org.apache.celeborn.common.rpc.RpcEnv;
import org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef;
import org.apache.celeborn.common.util.Utils;
+import org.apache.celeborn.common.util.Utils$;
import org.apache.celeborn.service.deploy.master.clustermeta.AbstractMetaManager;
public class RatisMasterStatusSystemSuiteJ {
@@ -62,95 +63,114 @@ public class RatisMasterStatusSystemSuiteJ {
resetRaftServer();
}
- public static void resetRaftServer() throws IOException, InterruptedException {
- Mockito.when(mockRpcEnv.setupEndpointRef(Mockito.any(), Mockito.any()))
- .thenReturn(mockRpcEndpoint);
- when(mockRpcEnv.setupEndpointRef(any(), any())).thenReturn(dummyRef);
-
+ private static void stopAllRaftServers() {
if (RATISSERVER1 != null) {
RATISSERVER1.stop();
}
-
if (RATISSERVER2 != null) {
RATISSERVER2.stop();
}
-
if (RATISSERVER3 != null) {
RATISSERVER3.stop();
}
+ }
- STATUSSYSTEM1 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
- STATUSSYSTEM2 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
- STATUSSYSTEM3 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
-
- MetaHandler handler1 = new MetaHandler(STATUSSYSTEM1);
- MetaHandler handler2 = new MetaHandler(STATUSSYSTEM2);
- MetaHandler handler3 = new MetaHandler(STATUSSYSTEM3);
-
- CelebornConf conf1 = new CelebornConf();
- File tmpDir1 = File.createTempFile("celeborn-ratis1", "for-test-only");
- tmpDir1.delete();
- tmpDir1.mkdirs();
- conf1.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir1.getAbsolutePath());
-
- CelebornConf conf2 = new CelebornConf();
- File tmpDir2 = File.createTempFile("celeborn-ratis2", "for-test-only");
- tmpDir2.delete();
- tmpDir2.mkdirs();
- conf2.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir2.getAbsolutePath());
-
- CelebornConf conf3 = new CelebornConf();
- File tmpDir3 = File.createTempFile("celeborn-ratis3", "for-test-only");
- tmpDir3.delete();
- tmpDir3.mkdirs();
- conf3.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir3.getAbsolutePath());
-
- String id1 = UUID.randomUUID().toString();
- String id2 = UUID.randomUUID().toString();
- String id3 = UUID.randomUUID().toString();
- int ratisPort1 = 9872;
- int ratisPort2 = 9873;
- int ratisPort3 = 9874;
-
- MasterNode masterNode1 =
- new MasterNode.Builder()
- .setHost(Utils.localHostName(conf1))
- .setRatisPort(ratisPort1)
- .setRpcPort(9872)
- .setNodeId(id1)
- .build();
- MasterNode masterNode2 =
- new MasterNode.Builder()
- .setHost(Utils.localHostName(conf2))
- .setRatisPort(ratisPort2)
- .setRpcPort(9873)
- .setNodeId(id2)
- .build();
- MasterNode masterNode3 =
- new MasterNode.Builder()
- .setHost(Utils.localHostName(conf3))
- .setRatisPort(ratisPort3)
- .setRpcPort(9874)
- .setNodeId(id3)
- .build();
-
- List<MasterNode> peersForNode1 = Arrays.asList(masterNode2, masterNode3);
- List<MasterNode> peersForNode2 = Arrays.asList(masterNode1, masterNode3);
- List<MasterNode> peersForNode3 = Arrays.asList(masterNode1, masterNode2);
-
- RATISSERVER1 = HARaftServer.newMasterRatisServer(handler1, conf1, masterNode1, peersForNode1);
- RATISSERVER2 = HARaftServer.newMasterRatisServer(handler2, conf2, masterNode2, peersForNode2);
- RATISSERVER3 = HARaftServer.newMasterRatisServer(handler3, conf3, masterNode3, peersForNode3);
-
- STATUSSYSTEM1.setRatisServer(RATISSERVER1);
- STATUSSYSTEM2.setRatisServer(RATISSERVER2);
- STATUSSYSTEM3.setRatisServer(RATISSERVER3);
-
- RATISSERVER1.start();
- RATISSERVER2.start();
- RATISSERVER3.start();
-
- Thread.sleep(15 * 1000);
+ public static void resetRaftServer() throws IOException, InterruptedException {
+ Mockito.when(mockRpcEnv.setupEndpointRef(Mockito.any(), Mockito.any()))
+ .thenReturn(mockRpcEndpoint);
+ when(mockRpcEnv.setupEndpointRef(any(), any())).thenReturn(dummyRef);
+
+ stopAllRaftServers();
+
+ int retryCount = 0;
+ boolean serversStarted = false;
+
+ while (!serversStarted) {
+ try {
+ STATUSSYSTEM1 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
+ STATUSSYSTEM2 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
+ STATUSSYSTEM3 = new HAMasterMetaManager(mockRpcEnv, new CelebornConf());
+
+ MetaHandler handler1 = new MetaHandler(STATUSSYSTEM1);
+ MetaHandler handler2 = new MetaHandler(STATUSSYSTEM2);
+ MetaHandler handler3 = new MetaHandler(STATUSSYSTEM3);
+
+ CelebornConf conf1 = new CelebornConf();
+ File tmpDir1 = File.createTempFile("celeborn-ratis1", "for-test-only");
+ tmpDir1.delete();
+ tmpDir1.mkdirs();
+ conf1.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir1.getAbsolutePath());
+
+ CelebornConf conf2 = new CelebornConf();
+ File tmpDir2 = File.createTempFile("celeborn-ratis2", "for-test-only");
+ tmpDir2.delete();
+ tmpDir2.mkdirs();
+ conf2.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir2.getAbsolutePath());
+
+ CelebornConf conf3 = new CelebornConf();
+ File tmpDir3 = File.createTempFile("celeborn-ratis3", "for-test-only");
+ tmpDir3.delete();
+ tmpDir3.mkdirs();
+ conf3.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR().key(), tmpDir3.getAbsolutePath());
+
+ String id1 = UUID.randomUUID().toString();
+ String id2 = UUID.randomUUID().toString();
+ String id3 = UUID.randomUUID().toString();
+
+ int ratisPort1 = Utils$.MODULE$.selectRandomPort(1024, 65535);
+ int ratisPort2 = ratisPort1 + 1;
+ int ratisPort3 = ratisPort2 + 1;
+
+ MasterNode masterNode1 =
+ new MasterNode.Builder()
+ .setHost(Utils.localHostName(conf1))
+ .setRatisPort(ratisPort1)
+ .setRpcPort(ratisPort1)
+ .setNodeId(id1)
+ .build();
+ MasterNode masterNode2 =
+ new MasterNode.Builder()
+ .setHost(Utils.localHostName(conf2))
+ .setRatisPort(ratisPort2)
+ .setRpcPort(ratisPort2)
+ .setNodeId(id2)
+ .build();
+ MasterNode masterNode3 =
+ new MasterNode.Builder()
+ .setHost(Utils.localHostName(conf3))
+ .setRatisPort(ratisPort3)
+ .setRpcPort(ratisPort3)
+ .setNodeId(id3)
+ .build();
+
+ List<MasterNode> peersForNode1 = Arrays.asList(masterNode2, masterNode3);
+ List<MasterNode> peersForNode2 = Arrays.asList(masterNode1, masterNode3);
+ List<MasterNode> peersForNode3 = Arrays.asList(masterNode1, masterNode2);
+
+ RATISSERVER1 =
+ HARaftServer.newMasterRatisServer(handler1, conf1, masterNode1, peersForNode1);
+ RATISSERVER2 =
+ HARaftServer.newMasterRatisServer(handler2, conf2, masterNode2, peersForNode2);
+ RATISSERVER3 =
+ HARaftServer.newMasterRatisServer(handler3, conf3, masterNode3, peersForNode3);
+
+ STATUSSYSTEM1.setRatisServer(RATISSERVER1);
+ STATUSSYSTEM2.setRatisServer(RATISSERVER2);
+ STATUSSYSTEM3.setRatisServer(RATISSERVER3);
+
+ RATISSERVER1.start();
+ RATISSERVER2.start();
+ RATISSERVER3.start();
+ Thread.sleep(15 * 1000);
+ serversStarted = true;
+ } catch (Exception e) {
+ stopAllRaftServers();
+ retryCount += 1;
+ if (retryCount == 3) {
+ throw e;
+ }
+ }
+ }
}
@Test
diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
index 46fbd67a6..1031dad1d 100644
--- a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
+++ b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
@@ -23,7 +23,7 @@ import org.scalatest.funsuite.AnyFunSuite
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.util.CelebornExitKind
+import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
class MasterSuite extends AnyFunSuite
with BeforeAndAfterAll
@@ -38,14 +38,16 @@ class MasterSuite extends AnyFunSuite
test("test single node startup functionality") {
val conf = new CelebornConf()
+ val randomMasterPort = Utils.selectRandomPort(1024, 65535)
+ val randomHttpPort = randomMasterPort + 1
conf.set(CelebornConf.HA_ENABLED.key, "false")
conf.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR.key, getTmpDir())
conf.set(CelebornConf.WORKER_STORAGE_DIRS.key, getTmpDir())
conf.set(CelebornConf.METRICS_ENABLED.key, "true")
conf.set(CelebornConf.MASTER_HTTP_HOST.key, "127.0.0.1")
- conf.set(CelebornConf.MASTER_HTTP_PORT.key, "11112")
+ conf.set(CelebornConf.MASTER_HTTP_PORT.key, randomHttpPort.toString)
- val args = Array("-h", "localhost", "-p", "9097")
+ val args = Array("-h", "localhost", "-p", randomMasterPort.toString)
val masterArgs = new MasterArguments(args, conf)
val master = new Master(conf, masterArgs)
diff --git a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/SplitTest.scala b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/SplitTest.scala
index e4ea37c15..1f0a312c2 100644
--- a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/SplitTest.scala
+++ b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/SplitTest.scala
@@ -31,15 +31,13 @@ import org.apache.celeborn.service.deploy.worker.Worker
class SplitTest extends AnyFunSuite with Logging with MiniClusterFeature
with BeforeAndAfterAll {
var workers: collection.Set[Worker] = null
+ var port = 0
override def beforeAll(): Unit = {
logInfo("test initialized , setup celeborn mini cluster")
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> "9097")
- val workerConf = Map(
- "celeborn.master.endpoints" -> "localhost:9097",
- CelebornConf.WORKER_FLUSHER_BUFFER_SIZE.key -> "10k")
- workers = setUpMiniCluster(masterConf, workerConf)._2
+ val (m, w) = setupMiniClusterWithRandomPorts(workerConf =
+ Map(CelebornConf.WORKER_FLUSHER_BUFFER_SIZE.key -> "10k"))
+ workers = w
+ port = m.conf.get(CelebornConf.MASTER_PORT)
}
override def afterAll(): Unit = {
logInfo("all test complete , stop celeborn mini cluster")
@@ -52,7 +50,7 @@ class SplitTest extends AnyFunSuite with Logging with MiniClusterFeature
configuration.setString(
"shuffle-service-factory.class",
"org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory")
- configuration.setString(CelebornConf.MASTER_ENDPOINTS.key, "localhost:9097")
+ configuration.setString(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$port")
configuration.setString("execution.batch-shuffle-mode", "ALL_EXCHANGES_BLOCKING")
configuration.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH)
configuration.setString("taskmanager.memory.network.min", "1024m")
diff --git a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
index fcb8a7b2b..a01251d09 100644
--- a/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
+++ b/tests/flink-it/src/test/scala/org/apache/celeborn/tests/flink/WordCountTest.scala
@@ -29,6 +29,7 @@ import org.apache.flink.streaming.api.graph.GlobalStreamExchangeMode
import org.scalatest.BeforeAndAfterAll
import org.scalatest.funsuite.AnyFunSuite
+import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.service.deploy.MiniClusterFeature
import org.apache.celeborn.service.deploy.worker.Worker
@@ -36,14 +37,13 @@ import org.apache.celeborn.service.deploy.worker.Worker
class WordCountTest extends AnyFunSuite with Logging with MiniClusterFeature
with BeforeAndAfterAll {
var workers: collection.Set[Worker] = null
+ var port = 0
override def beforeAll(): Unit = {
logInfo("test initialized , setup celeborn mini cluster")
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> "9097")
- val workerConf = Map("celeborn.master.endpoints" -> "localhost:9097")
- workers = setUpMiniCluster(masterConf, workerConf)._2
+ val (m, w) = setupMiniClusterWithRandomPorts()
+ workers = w
+ port = m.conf.get(CelebornConf.MASTER_PORT)
}
override def afterAll(): Unit = {
@@ -58,7 +58,7 @@ class WordCountTest extends AnyFunSuite with Logging with MiniClusterFeature
configuration.setString(
"shuffle-service-factory.class",
"org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory")
- configuration.setString("celeborn.master.endpoints", "localhost:9097")
+ configuration.setString("celeborn.master.endpoints", s"localhost:$port")
configuration.setString("execution.batch-shuffle-mode", "ALL_EXCHANGES_BLOCKING")
configuration.set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH)
configuration.setString("taskmanager.memory.network.min", "1024m")
diff --git a/tests/mr-it/src/test/scala/org/apache/celeborn/tests/mr/WordCountTest.scala b/tests/mr-it/src/test/scala/org/apache/celeborn/tests/mr/WordCountTest.scala
index 0f1bb91aa..d7c0ca41f 100644
--- a/tests/mr-it/src/test/scala/org/apache/celeborn/tests/mr/WordCountTest.scala
+++ b/tests/mr-it/src/test/scala/org/apache/celeborn/tests/mr/WordCountTest.scala
@@ -37,27 +37,29 @@ import org.apache.hadoop.yarn.server.MiniYARNCluster
import org.scalatest.BeforeAndAfterAll
import org.scalatest.funsuite.AnyFunSuite
+import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.util.Utils
import org.apache.celeborn.service.deploy.MiniClusterFeature
+import org.apache.celeborn.service.deploy.master.Master
import org.apache.celeborn.service.deploy.worker.Worker
class WordCountTest extends AnyFunSuite with Logging with MiniClusterFeature
with BeforeAndAfterAll {
var workers: collection.Set[Worker] = null
+ var master: Master = null
var yarnCluster: MiniYARNCluster = null
var hadoopConf: Configuration = null
override def beforeAll(): Unit = {
logInfo("test initialized , setup celeborn mini cluster")
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> "9097")
- val workerConf = Map("celeborn.master.endpoints" -> "localhost:9097")
- workers = setUpMiniCluster(masterConf, workerConf)._2
+ val (newMaster, newWorkers) = setupMiniClusterWithRandomPorts()
+ master = newMaster
+ workers = newWorkers
hadoopConf = new Configuration()
+ hadoopConf.setInt("yarn.resourcemanager.am.max-attempts", 1)
hadoopConf.set("yarn.scheduler.capacity.root.queues", "default,other_queue")
hadoopConf.setInt("yarn.scheduler.capacity.root.default.capacity", 100)
@@ -99,47 +101,72 @@ class WordCountTest extends AnyFunSuite with Logging with MiniClusterFeature
val output = Utils.createTempDir(System.getProperty("java.io.tmpdir"), "output")
val mrOutputPath = new Path(output.getPath + File.separator + "mr_output")
- val conf = new Configuration(yarnCluster.getConfig)
- // YARN config
- conf.set("yarn.app.mapreduce.am.job.recovery.enable", "false")
- conf.set(
- "yarn.app.mapreduce.am.command-opts",
- "org.apache.celeborn.mapreduce.v2.app.MRAppMasterWithCeleborn")
-
- // MapReduce config
- conf.set("mapreduce.framework.name", "yarn")
- conf.set("mapreduce.job.user.classpath.first", "true")
-
- conf.set("mapreduce.job.reduce.slowstart.completedmaps", "1")
- conf.set("mapreduce.celeborn.master.endpoints", "localhost:9097")
- conf.set(
- MRJobConfig.MAP_OUTPUT_COLLECTOR_CLASS_ATTR,
- "org.apache.hadoop.mapred.CelebornMapOutputCollector")
- conf.set(
- "mapreduce.job.reduce.shuffle.consumer.plugin.class",
- "org.apache.hadoop.mapreduce.task.reduce.CelebornShuffleConsumer")
-
- val job = Job.getInstance(conf, "word count")
- job.setJarByClass(classOf[WordCount])
- job.setMapperClass(classOf[WordCount.TokenizerMapper])
- job.setCombinerClass(classOf[WordCount.IntSumReducer])
- job.setReducerClass(classOf[WordCount.IntSumReducer])
- job.setOutputKeyClass(classOf[Text])
- job.setOutputValueClass(classOf[IntWritable])
- FileInputFormat.addInputPath(job, new Path(input.getPath))
- FileOutputFormat.setOutputPath(job, mrOutputPath)
-
- val mapreduceLibPath = (Utils.getCodeSourceLocation(getClass).split("/").dropRight(1) ++ Array(
- "mapreduce_lib")).mkString("/")
- val excludeJarList =
- Seq("hadoop-client-api", "hadoop-client-runtime", "hadoop-client-minicluster")
- Files.list(Paths.get(mapreduceLibPath)).iterator().asScala.foreach(path => {
- if (!excludeJarList.exists(path.toFile.getPath.contains(_))) {
- job.addFileToClassPath(new Path(path.toString))
+ var finish = false
+ var exitCode = false
+ var retryCount = 0
+ while (!finish) {
+ try {
+ val conf = new Configuration(yarnCluster.getConfig)
+ // YARN config
+ conf.set("yarn.app.mapreduce.am.job.recovery.enable", "false")
+ conf.set(
+ "yarn.app.mapreduce.am.command-opts",
+ "org.apache.celeborn.mapreduce.v2.app.MRAppMasterWithCeleborn")
+
+ // MapReduce config
+ conf.set("mapreduce.framework.name", "yarn")
+ conf.set("mapreduce.job.user.classpath.first", "true")
+
+ conf.set("mapreduce.job.reduce.slowstart.completedmaps", "1")
+ conf.set(
+ "mapreduce.celeborn.master.endpoints",
+ s"localhost:${master.conf.get(CelebornConf.MASTER_PORT)}")
+ conf.set(
+ MRJobConfig.MAP_OUTPUT_COLLECTOR_CLASS_ATTR,
+ "org.apache.hadoop.mapred.CelebornMapOutputCollector")
+ conf.set(
+ "mapreduce.job.reduce.shuffle.consumer.plugin.class",
+ "org.apache.hadoop.mapreduce.task.reduce.CelebornShuffleConsumer")
+
+ val job = Job.getInstance(conf, "word count")
+ job.setJarByClass(classOf[WordCount])
+ job.setMapperClass(classOf[WordCount.TokenizerMapper])
+ job.setCombinerClass(classOf[WordCount.IntSumReducer])
+ job.setReducerClass(classOf[WordCount.IntSumReducer])
+ job.setOutputKeyClass(classOf[Text])
+ job.setOutputValueClass(classOf[IntWritable])
+ FileInputFormat.addInputPath(job, new Path(input.getPath))
+ FileOutputFormat.setOutputPath(job, mrOutputPath)
+
+ val mapreduceLibPath =
+ (Utils.getCodeSourceLocation(getClass).split("/").dropRight(1) ++ Array(
+ "mapreduce_lib")).mkString("/")
+ val excludeJarList =
+ Seq("hadoop-client-api", "hadoop-client-runtime", "hadoop-client-minicluster")
+ Files.list(Paths.get(mapreduceLibPath)).iterator().asScala.foreach(path => {
+ if (!excludeJarList.exists(path.toFile.getPath.contains(_))) {
+ job.addFileToClassPath(new Path(path.toString))
+ }
+ })
+
+ exitCode = job.waitForCompletion(true)
+ if (exitCode) {
+ finish = true
+ } else {
+ retryCount += 1
+ if (retryCount >= 2) {
+ throw new RuntimeException("failed to run wordcount")
+ }
+ }
+ } catch {
+ case e: Exception =>
+ retryCount += 1
+ if (retryCount >= 2) {
+ log.error("failed to run wordcount", e)
+ throw e
+ }
}
- })
-
- val exitCode = job.waitForCompletion(true)
+ }
assert(exitCode, "Returned error code.")
val outputFilePath = Paths.get(mrOutputPath.toString, "part-r-00000")
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerCommitFilesSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerCommitFilesSuite.scala
index 23452bafa..0d73d88b8 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerCommitFilesSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerCommitFilesSuite.scala
@@ -33,20 +33,17 @@ import org.apache.celeborn.service.deploy.MiniClusterFeature
import org.apache.celeborn.service.deploy.worker.CommitInfo
class LifecycleManagerCommitFilesSuite extends WithShuffleClientSuite with MiniClusterFeature {
- private val masterPort = 19097
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
override def beforeAll(): Unit = {
super.beforeAll()
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
- setUpMiniCluster(masterConf, workerConf)
+ val (master, _) = setupMiniClusterWithRandomPorts()
+ celebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ master.conf.get(CelebornConf.MASTER_ENDPOINTS.key))
}
test("test commit files without mocking failure") {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerDestroySlotsSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerDestroySlotsSuite.scala
index bc9b28484..1fc639d60 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerDestroySlotsSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerDestroySlotsSuite.scala
@@ -30,20 +30,17 @@ import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
import org.apache.celeborn.service.deploy.MiniClusterFeature
class LifecycleManagerDestroySlotsSuite extends WithShuffleClientSuite with MiniClusterFeature {
- private val masterPort = 19097
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
override def beforeAll(): Unit = {
super.beforeAll()
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
- setUpMiniCluster(masterConf, workerConf)
+ val (master, _) = setupMiniClusterWithRandomPorts()
+ celebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ master.conf.get(CelebornConf.MASTER_ENDPOINTS.key))
}
test("test destroy workers without mocking failure") {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSetupEndpointSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSetupEndpointSuite.scala
index 3c0a4ba22..8fee9cbb9 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSetupEndpointSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSetupEndpointSuite.scala
@@ -31,18 +31,16 @@ import org.apache.celeborn.service.deploy.MiniClusterFeature
class LifecycleManagerSetupEndpointSuite extends WithShuffleClientSuite with MiniClusterFeature {
private val masterPort = 19097
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
override def beforeAll(): Unit = {
super.beforeAll()
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
- setUpMiniCluster(masterConf, workerConf)
+ val (master, _) = setupMiniClusterWithRandomPorts()
+ celebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ master.conf.get(CelebornConf.MASTER_ENDPOINTS.key))
}
test("test setup endpoints with all workers good") {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
index d310011c8..1bcedc75f 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
@@ -19,26 +19,26 @@ package org.apache.celeborn.tests.client
import java.util
+import scala.collection.JavaConverters._
+
import org.apache.celeborn.client.{LifecycleManager, WithShuffleClientSuite}
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.protocol.message.StatusCode
import org.apache.celeborn.service.deploy.MiniClusterFeature
class LifecycleManagerSuite extends WithShuffleClientSuite with MiniClusterFeature {
- private val masterPort = 19097
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
override def beforeAll(): Unit = {
super.beforeAll()
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
- setUpMiniCluster(masterConf, workerConf)
+ val (master, _) = setupMiniClusterWithRandomPorts()
+ logInfo(s"master address is: ${master.conf.get(CelebornConf.MASTER_ENDPOINTS.key)}")
+ celebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ master.conf.get(CelebornConf.MASTER_ENDPOINTS.key))
}
test("CELEBORN-1151: test request slots with client blacklist worker with filter enabled") {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
index 4946a3dae..112174fbf 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ShuffleClientSuite.scala
@@ -26,28 +26,31 @@ import org.apache.celeborn.service.deploy.MiniClusterFeature
class ShuffleClientSuite extends WithShuffleClientSuite with MiniClusterFeature {
private val masterPort = 19097
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ celebornConf
.set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
override def beforeAll(): Unit = {
super.beforeAll()
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
- setUpMiniCluster(masterConf, workerConf)
+ val (master, _) = setupMiniClusterWithRandomPorts()
+ celebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ master.conf.get(CelebornConf.MASTER_ENDPOINTS.key))
+ celebornConf.set(
+ CelebornConf.MASTER_PORT.key,
+ master.conf.get(CelebornConf.MASTER_PORT.key))
}
test("test register when master not available") {
- val celebornConf: CelebornConf = new CelebornConf()
- celebornConf.set(CelebornConf.MASTER_ENDPOINTS.key, "localhost:19098")
- celebornConf.set(CelebornConf.MASTER_CLIENT_MAX_RETRIES.key, "0")
+ val newCelebornConf: CelebornConf = new CelebornConf()
+ newCelebornConf.set(
+ CelebornConf.MASTER_ENDPOINTS.key,
+ s"localhost:${celebornConf.get(CelebornConf.MASTER_PORT) + 1}")
+ newCelebornConf.set(CelebornConf.MASTER_CLIENT_MAX_RETRIES.key, "0")
- val lifecycleManager: LifecycleManager = new LifecycleManager(APP, celebornConf)
+ val lifecycleManager: LifecycleManager = new LifecycleManager(APP, newCelebornConf)
val shuffleClient: ShuffleClientImpl = {
- val client = new ShuffleClientImpl(APP, celebornConf, userIdentifier)
+ val client = new ShuffleClientImpl(APP, newCelebornConf, userIdentifier)
client.setupLifecycleManagerRef(lifecycleManager.self)
client
}
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
index 8983f6bd6..fe9683c37 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
@@ -48,7 +48,9 @@ class CelebornFetchFailureSuite extends AnyFunSuite
override def createWorker(map: Map[String, String]): Worker = {
val storageDir = createTmpDir()
- workerDirs = workerDirs :+ storageDir
+ this.synchronized {
+ workerDirs = workerDirs :+ storageDir
+ }
super.createWorker(map, storageDir)
}
@@ -75,14 +77,15 @@ class CelebornFetchFailureSuite extends AnyFunSuite
h.userIdentifier,
h.extension)
val celebornShuffleId = SparkUtils.celebornShuffleId(shuffleClient, h, context, false)
- val datafile =
- workerDirs.map(dir => {
- new File(s"$dir/celeborn-worker/shuffle_data/$appUniqueId/$celebornShuffleId")
- }).filter(_.exists())
- .flatMap(_.listFiles().iterator).headOption
+ val allFiles = workerDirs.map(dir => {
+ new File(s"$dir/celeborn-worker/shuffle_data/$appUniqueId/$celebornShuffleId")
+ })
+ val datafile = allFiles.filter(_.exists())
+ .flatMap(_.listFiles().iterator).headOption
datafile match {
case Some(file) => file.delete()
- case None => throw new RuntimeException("unexpected, there must be some data file")
+ case None => throw new RuntimeException("unexpected, there must be some data file" +
+ s" under ${workerDirs.mkString(",")}")
}
}
case _ => throw new RuntimeException("unexpected, only support RssShuffleHandle here")
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
index c2a7f1167..9fe49bfd4 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornHashCheckDiskSuite.scala
@@ -39,7 +39,7 @@ class CelebornHashCheckDiskSuite extends SparkTestBase {
val workerConf = Map(
CelebornConf.WORKER_STORAGE_DIRS.key -> "/tmp:capacity=1000",
CelebornConf.WORKER_HEARTBEAT_TIMEOUT.key -> "10s")
- workers = setUpMiniCluster(masterConf, workerConf)._2
+ workers = setupMiniClusterWithRandomPorts(masterConf, workerConf)._2
}
override def beforeEach(): Unit = {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
index 387792da4..5bf7c1303 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
@@ -44,7 +44,7 @@ class PushDataTimeoutTest extends AnyFunSuite
// enabled, there is a possibility that two workers might be added to the excluded list due to
// primary/replica timeout issues, then there are not enough workers to do replication if
// available workers number = 1
- setUpMiniCluster(masterConf = null, workerConf = workerConf, workerNum = 4)
+ setupMiniClusterWithRandomPorts(workerConf = workerConf, workerNum = 4)
}
override def beforeEach(): Unit = {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/RetryReviveTest.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/RetryReviveTest.scala
index 280c16575..ea6eed601 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/RetryReviveTest.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/RetryReviveTest.scala
@@ -32,7 +32,7 @@ class RetryReviveTest extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized , setup celeborn mini cluster")
- setUpMiniCluster(masterConf = null)
+ setupMiniClusterWithRandomPorts()
}
override def beforeEach(): Unit = {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
index cbf973b92..059e7d033 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/ShuffleFallbackSuite.scala
@@ -51,7 +51,7 @@ class ShuffleFallbackSuite extends AnyFunSuite
}
test(s"celeborn spark integration test - fallback") {
- setUpMiniCluster(workerNum = 5)
+ setupMiniClusterWithRandomPorts(workerNum = 5)
val sparkConf = new SparkConf().setAppName("celeborn-demo")
.setMaster("local[2]")
.set(s"spark.${CelebornConf.SPARK_SHUFFLE_FORCE_FALLBACK_ENABLED.key}", "true")
@@ -67,7 +67,7 @@ class ShuffleFallbackSuite extends AnyFunSuite
}
test(s"celeborn spark integration test - fallback with workers unavailable") {
- setUpMiniCluster(workerNum = 0)
+ setupMiniClusterWithRandomPorts(workerNum = 0)
val sparkConf = new SparkConf().setAppName("celeborn-demo")
.setMaster("local[2]")
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersLargeTest.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersLargeTest.scala
index 4114e13c1..b5cca2d22 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersLargeTest.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersLargeTest.scala
@@ -32,9 +32,8 @@ class SlotsAssignMaxWorkersLargeTest extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized, setup Celeborn mini cluster")
- val masterConf = Map(
- s"${CelebornConf.CLIENT_SLOT_ASSIGN_MAX_WORKERS.key}" -> "10")
- setUpMiniCluster(masterConf = masterConf, workerConf = null)
+ setupMiniClusterWithRandomPorts(masterConf = Map(
+ s"${CelebornConf.CLIENT_SLOT_ASSIGN_MAX_WORKERS.key}" -> "10"))
}
override def beforeEach(): Unit = {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersSmallTest.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersSmallTest.scala
index 20c33ae0e..fab1f7448 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersSmallTest.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SlotsAssignMaxWorkersSmallTest.scala
@@ -32,9 +32,8 @@ class SlotsAssignMaxWorkersSmallTest extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized, setup Celeborn mini cluster")
- val masterConf = Map(
- s"${CelebornConf.CLIENT_SLOT_ASSIGN_MAX_WORKERS.key}" -> "5")
- setUpMiniCluster(masterConf = masterConf, workerConf = null)
+ setupMiniClusterWithRandomPorts(masterConf = Map(
+ s"${CelebornConf.CLIENT_SLOT_ASSIGN_MAX_WORKERS.key}" -> "5"))
}
override def beforeEach(): Unit = {
diff --git a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
index e2cb3c98a..552fe3354 100644
--- a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
+++ b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/SparkTestBase.scala
@@ -39,7 +39,7 @@ trait SparkTestBase extends AnyFunSuite
override def beforeAll(): Unit = {
logInfo("test initialized , setup Celeborn mini cluster")
- setUpMiniCluster(workerNum = 5)
+ setupMiniClusterWithRandomPorts(workerNum = 5)
}
override def afterAll(): Unit = {
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index 6dd9d48d1..eafc3e5a7 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -852,6 +852,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
fileMeta,
filePath,
StorageInfo.Type.HDD)
+ logInfo(s"created file at $filePath")
diskFileInfos.computeIfAbsent(shuffleKey, diskFileInfoMapFunc).put(
fileName,
diskFileInfo)
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/HeartbeatFeature.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/HeartbeatFeature.scala
index cf216a429..dc5616064 100644
--- a/worker/src/test/scala/org/apache/celeborn/service/deploy/HeartbeatFeature.scala
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/HeartbeatFeature.scala
@@ -32,13 +32,11 @@ trait HeartbeatFeature extends MiniClusterFeature {
dataClientFactory: TransportClientFactory,
assertFunc: (TransportClient, TransportClient) => Unit): Unit = {
logInfo("test initialized , setup celeborn mini cluster")
- val masterConf = Map(
- CelebornConf.MASTER_HOST.key -> "localhost",
- CelebornConf.MASTER_PORT.key -> "9097")
var master: Master = null
var workers: collection.Set[Worker] = null
try {
- val (_master, _workers) = setUpMiniCluster(masterConf, workerConf, workerNum = 1)
+ val (_master, _workers) =
+ setupMiniClusterWithRandomPorts(workerConf = workerConf, workerNum = 1)
master = _master
workers = _workers
workers.foreach { w =>
@@ -62,7 +60,6 @@ trait HeartbeatFeature extends MiniClusterFeature {
def getTestHeartbeatFromWorker2ClientConf: (Map[String, String], CelebornConf) = {
val workerConf = Map(
- CelebornConf.MASTER_ENDPOINTS.key -> "localhost:9097",
"celeborn.push.heartbeat.interval" -> "4s",
"celeborn.worker.push.heartbeat.enabled" -> "true",
"celeborn.worker.fetch.heartbeat.enabled" -> "true",
@@ -90,7 +87,6 @@ trait HeartbeatFeature extends MiniClusterFeature {
def getTestHeartbeatFromWorker2ClientWithNoHeartbeatConf: (Map[String, String], CelebornConf) = {
val workerConf = Map(
- "celeborn.master.endpoints" -> "localhost:9097",
"celeborn.push.heartbeat.interval" -> "4s",
"celeborn.fetch.heartbeat.interval" -> "4s",
"celeborn.worker.push.heartbeat.enabled" -> "false",
@@ -119,7 +115,6 @@ trait HeartbeatFeature extends MiniClusterFeature {
def getTestHeartbeatFromWorker2ClientWithCloseChannelConf: (Map[String, String], CelebornConf) = {
val workerConf = Map(
- CelebornConf.MASTER_ENDPOINTS.key -> "localhost:9097",
"celeborn.fetch.io.connectionTimeout" -> "9s",
"celeborn.push.io.connectionTimeout" -> "9s",
"celeborn.push.heartbeat.interval" -> "4s",
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
index 8c8411910..beecb76a5 100644
--- a/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/MiniClusterFeature.scala
@@ -17,29 +17,70 @@
package org.apache.celeborn.service.deploy
+import java.net.BindException
import java.nio.file.Files
-import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.locks.{Lock, ReentrantLock}
import scala.collection.mutable
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
+import org.apache.celeborn.common.util.Utils.selectRandomPort
import org.apache.celeborn.service.deploy.master.{Master, MasterArguments}
import org.apache.celeborn.service.deploy.worker.{Worker, WorkerArguments}
import org.apache.celeborn.service.deploy.worker.memory.MemoryManager
trait MiniClusterFeature extends Logging {
- val masterHttpPort = new AtomicInteger(22378)
- val workerHttpPort = new AtomicInteger(12378)
+
var masterInfo: (Master, Thread) = _
val workerInfos = new mutable.HashMap[Worker, Thread]()
- def runnerWrap[T](code: => T): Thread = new Thread(new Runnable {
+ class RunnerWrap[T](code: => T) extends Thread {
+
override def run(): Unit = {
Utils.tryLogNonFatalError(code)
}
- })
+ }
+
+ def setupMiniClusterWithRandomPorts(
+ masterConf: Map[String, String] = Map(),
+ workerConf: Map[String, String] = Map(),
+ workerNum: Int = 3): (Master, collection.Set[Worker]) = {
+ var retryCount = 0
+ var created = false
+ var master: Master = null
+ var workers: collection.Set[Worker] = null
+ while (!created) {
+ try {
+ val randomPort = selectRandomPort(1024, 65535)
+ val finalMasterConf = Map(
+ s"${CelebornConf.MASTER_HOST.key}" -> "localhost",
+ s"${CelebornConf.PORT_MAX_RETRY.key}" -> "0",
+ s"${CelebornConf.MASTER_PORT.key}" -> s"$randomPort",
+ s"${CelebornConf.MASTER_ENDPOINTS.key}" -> s"localhost:$randomPort") ++
+ masterConf
+ val finalWorkerConf = Map(
+ s"${CelebornConf.MASTER_ENDPOINTS.key}" -> s"localhost:$randomPort") ++
+ workerConf
+ logInfo(s"generated configuration $finalMasterConf")
+ val (m, w) =
+ setUpMiniCluster(masterConf = finalMasterConf, workerConf = finalWorkerConf, workerNum)
+ master = m
+ workers = w
+ created = true
+ } catch {
+ case e: BindException =>
+ logError(s"failed to setup mini cluster, retrying (retry count: $retryCount)", e)
+ retryCount += 1
+ if (retryCount == 3) {
+ logError("failed to setup mini cluster, reached the max retry count", e)
+ throw e
+ }
+ }
+ }
+ (master, workers)
+ }
def createTmpDir(): String = {
val tmpDir = Files.createTempDirectory("celeborn-")
@@ -51,7 +92,7 @@ trait MiniClusterFeature extends Logging {
private def createMaster(map: Map[String, String] = null): Master = {
val conf = new CelebornConf()
conf.set(CelebornConf.METRICS_ENABLED.key, "false")
- val httpPort = masterHttpPort.getAndIncrement()
+ val httpPort = selectRandomPort(1024, 65535)
conf.set(CelebornConf.MASTER_HTTP_PORT.key, s"$httpPort")
logInfo(s"set ${CelebornConf.MASTER_HTTP_PORT.key} to $httpPort")
if (map != null) {
@@ -76,7 +117,7 @@ trait MiniClusterFeature extends Logging {
conf.set(CelebornConf.WORKER_STORAGE_DIRS.key, storageDir)
conf.set(CelebornConf.WORKER_DISK_MONITOR_ENABLED.key, "false")
conf.set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
- conf.set(CelebornConf.WORKER_HTTP_PORT.key, s"${workerHttpPort.incrementAndGet()}")
+ conf.set(CelebornConf.WORKER_HTTP_PORT.key, s"${selectRandomPort(1024, 65535)}")
conf.set("celeborn.fetch.io.threads", "4")
conf.set("celeborn.push.io.threads", "4")
if (map != null) {
@@ -98,24 +139,97 @@ trait MiniClusterFeature extends Logging {
}
}
- def setUpMiniCluster(
+ private def setUpMiniCluster(
masterConf: Map[String, String] = null,
workerConf: Map[String, String] = null,
workerNum: Int = 3): (Master, collection.Set[Worker]) = {
+ val timeout = 30000
val master = createMaster(masterConf)
- val masterThread = runnerWrap(master.rpcEnv.awaitTermination())
+ val masterStartedSignal = Array(false)
+ val masterThread = new RunnerWrap({
+ try {
+ masterStartedSignal(0) = true
+ master.rpcEnv.awaitTermination()
+ } catch {
+ case ex: Exception =>
+ masterStartedSignal(0) = false
+ throw ex
+ }
+ })
masterThread.start()
masterInfo = (master, masterThread)
- Thread.sleep(5000L)
- (1 to workerNum).foreach { _ =>
- val worker = createWorker(workerConf)
- val workerThread = runnerWrap(worker.initialize())
- workerThread.start()
- workerInfos.put(worker, workerThread)
+ var masterStartWaitingTime = 0
+ Thread.sleep(5000)
+ while (!masterStartedSignal.head) {
+ logInfo("waiting for master node starting")
+ masterStartWaitingTime += 5000
+ if (masterStartWaitingTime >= timeout) {
+ throw new BindException("cannot start master rpc endpoint")
+ }
}
- Thread.sleep(5000L)
- workerInfos.foreach { case (worker, _) => assert(worker.registered.get()) }
+ val workers = new Array[Worker](workerNum)
+ val flagUpdateLock = new ReentrantLock()
+ val threads = (1 to workerNum).map { i =>
+ val workerThread = new RunnerWrap({
+ var workerStartRetry = 0
+ var workerStarted = false
+ while (!workerStarted) {
+ try {
+ val worker = createWorker(workerConf)
+ flagUpdateLock.lock()
+ workers(i - 1) = worker
+ flagUpdateLock.unlock()
+ workerStarted = true
+ worker.initialize()
+ } catch {
+ case ex: Exception =>
+ if (workers(i - 1) != null) {
+ workers(i - 1).shutdownGracefully()
+ }
+ workerStarted = false
+ workerStartRetry += 1
+ logError(s"cannot start worker $i, retrying: ", ex)
+ if (workerStartRetry == 3) {
+ logError(s"cannot start worker $i, reached to max retrying", ex)
+ throw ex
+ } else {
+ Thread.sleep(math.pow(2000, workerStartRetry).toInt)
+ }
+ }
+ }
+ })
+ workerThread.setName(s"worker ${i} starter thread")
+ workerThread
+ }
+ threads.foreach(_.start())
+ Thread.sleep(5000)
+ var allWorkersStarted = false
+ var workersWaitingTime = 0
+ while (!allWorkersStarted) {
+ try {
+ (0 until workerNum).foreach { i =>
+ {
+ if (workers(i) == null) {
+ throw new IllegalStateException(s"worker $i hasn't been initialized")
+ } else {
+ workerInfos.put(workers(i), threads(i))
+ }
+ }
+ }
+ workerInfos.foreach { case (worker, _) => assert(worker.registered.get()) }
+ allWorkersStarted = true
+ } catch {
+ case ex: Exception =>
+ logError("all workers haven't been started retrying", ex)
+ Thread.sleep(5000)
+ workersWaitingTime += 5000
+ if (workersWaitingTime >= timeout) {
+ logError(s"cannot start all workers after $timeout ms", ex)
+ throw ex
+ }
+ }
+ }
(master, workerInfos.keySet)
}
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/ReadWriteTestBase.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/ReadWriteTestBase.scala
index f8cbec595..3009a6445 100644
--- a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/ReadWriteTestBase.scala
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/ReadWriteTestBase.scala
@@ -38,16 +38,13 @@ import org.apache.celeborn.service.deploy.MiniClusterFeature
trait ReadWriteTestBase extends AnyFunSuite
with Logging with MiniClusterFeature with BeforeAndAfterAll {
- val masterPort = 19097
+
+ var masterPort = 0
override def beforeAll(): Unit = {
- val masterConf = Map(
- "celeborn.master.host" -> "localhost",
- "celeborn.master.port" -> masterPort.toString)
- val workerConf = Map(
- "celeborn.master.endpoints" -> s"localhost:$masterPort")
logInfo("test initialized , setup Celeborn mini cluster")
- setUpMiniCluster(masterConf, workerConf)
+ val (m, _) = setupMiniClusterWithRandomPorts()
+ masterPort = m.conf.masterPort
}
override def afterAll(): Unit = {