This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new c799373793 [CELEBORN-2275][CIP-14] Add C++ merge-write and Java-read
hybrid integration test
c799373793 is described below
commit c79937379383b305d43d275eb1e2b9e73d9ab439
Author: afterincomparableyum
<[email protected]>
AuthorDate: Tue Jun 2 19:47:04 2026 +0800
[CELEBORN-2275][CIP-14] Add C++ merge-write and Java-read hybrid
integration test
### What changes were proposed in this pull request?
Add a new C++ test client that for the mergeData/pushMergedData write path
and validates data integrity by reading back from the Java ShuffleClient. This
complements the existing pushData based hybrid test by covering the merge write
path.
- Add DataSumWithMergeWriterClient.cpp and its CMake build target
- Add CppMergeWriteJavaReadTest entry points for NONE, LZ4, and ZSTD
compression codecs
- Add runCppMergeWriteJavaRead to JavaCppHybridReadWriteTestBase
- Update cpp_integration CI workflow to run the new tests
### Why are the changes needed?
This is to add integration tests for
https://github.com/apache/celeborn/pull/3611.
### Does this PR resolve a correctness bug?
No.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Tested through running unit tests and compiling locally.
Closes #3619 from afterincomparableyum/cpp-client/celeborn-2275.
Lead-authored-by: afterincomparableyum
<[email protected]>
Co-authored-by: afterincomparableyum <afterincomparableyum>
Signed-off-by: SteNicholas <[email protected]>
---
.github/workflows/cpp_integration.yml | 48 ++-
.../org/apache/celeborn/common/util/Utils.scala | 5 +
cpp/celeborn/tests/CMakeLists.txt | 38 +-
...Client.cpp => DataSumWithMergeWriterClient.cpp} | 32 +-
cpp/celeborn/tests/DataSumWithReaderClient.cpp | 2 +
cpp/celeborn/tests/DataSumWithWriterClient.cpp | 2 +
.../cluster/CppMergeWriteJavaReadTestWithLZ4.scala | 27 ++
.../CppMergeWriteJavaReadTestWithNONE.scala | 27 ++
.../CppMergeWriteJavaReadTestWithZSTD.scala | 27 ++
.../cluster/JavaCppHybridReadWriteTestBase.scala | 391 ++++++++++++++-------
10 files changed, 413 insertions(+), 186 deletions(-)
diff --git a/.github/workflows/cpp_integration.yml
b/.github/workflows/cpp_integration.yml
index 804f98c820..ce288a9232 100644
--- a/.github/workflows/cpp_integration.yml
+++ b/.github/workflows/cpp_integration.yml
@@ -87,43 +87,55 @@ jobs:
run: build/mvn clean install -DskipTests
- name: Run Java-Write Cpp-Read Hybrid Integration Test (NONE
Decompression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithNONE"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithNONE"
- name: Run Java-Write Cpp-Read Hybrid Integration Test (LZ4
Decompression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithLZ4"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithLZ4"
- name: Run Java-Write Cpp-Read Hybrid Integration Test (ZSTD
Decompression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithZSTD"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.JavaWriteCppReadTestWithZSTD"
- name: Run Cpp-Write Java-Read Hybrid Integration Test (NONE
Compression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithNONE"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithNONE"
- name: Run Cpp-Write Java-Read Hybrid Integration Test (LZ4 Compression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithLZ4"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithLZ4"
- name: Run Cpp-Write Java-Read Hybrid Integration Test (ZSTD
Compression)
run: |
- build/mvn -pl worker \
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
test-compile exec:java \
-Dexec.classpathScope="test" \
-
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithZSTD"
\
- -Dexec.args="-XX:MaxDirectMemorySize=2G"
\ No newline at end of file
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppWriteJavaReadTestWithZSTD"
+ - name: Run Cpp-MergeWrite Java-Read Hybrid Integration Test (NONE
Compression)
+ run: |
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
+ test-compile exec:java \
+ -Dexec.classpathScope="test" \
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppMergeWriteJavaReadTestWithNONE"
+ - name: Run Cpp-MergeWrite Java-Read Hybrid Integration Test (LZ4
Compression)
+ run: |
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
+ test-compile exec:java \
+ -Dexec.classpathScope="test" \
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppMergeWriteJavaReadTestWithLZ4"
+ - name: Run Cpp-MergeWrite Java-Read Hybrid Integration Test (ZSTD
Compression)
+ run: |
+ MAVEN_OPTS="-XX:MaxDirectMemorySize=2G" build/mvn -pl worker \
+ test-compile exec:java \
+ -Dexec.classpathScope="test" \
+
-Dexec.mainClass="org.apache.celeborn.service.deploy.cluster.CppMergeWriteJavaReadTestWithZSTD"
\ No newline at end of file
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 9c973e324d..5b2dd6a109 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
@@ -834,6 +834,11 @@ object Utils extends Logging {
readProcessStdout(process)
}
+ def runCommand(cmd: Seq[String]): String = {
+ val process = Runtime.getRuntime.exec(cmd.toArray)
+ readProcessStdout(process)
+ }
+
def runCommandComplex(cmd: String): String = {
val cmds = Array("/bin/sh", "-c", cmd)
val process = Runtime.getRuntime.exec(cmds)
diff --git a/cpp/celeborn/tests/CMakeLists.txt
b/cpp/celeborn/tests/CMakeLists.txt
index 104607ce2f..77f88874fe 100644
--- a/cpp/celeborn/tests/CMakeLists.txt
+++ b/cpp/celeborn/tests/CMakeLists.txt
@@ -12,12 +12,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-add_library(
- dataSumWithReaderClient
- DataSumWithReaderClient.cpp)
-
-target_link_libraries(
- dataSumWithReaderClient
+set(DATA_SUM_CLIENT_DEPS
memory
utils
conf
@@ -30,34 +25,13 @@ target_link_libraries(
${LIBSODIUM_LIBRARY}
${FOLLY_WITH_DEPENDENCIES}
${GLOG}
- ${GFLAGS_LIBRARIES}
-)
+ ${GFLAGS_LIBRARIES})
add_executable(cppDataSumWithReaderClient DataSumWithReaderClient.cpp)
-
-target_link_libraries(cppDataSumWithReaderClient dataSumWithReaderClient)
-
-add_library(
- dataSumWithWriterClient
- DataSumWithWriterClient.cpp)
-
-target_link_libraries(
- dataSumWithWriterClient
- memory
- utils
- conf
- proto
- network
- protocol
- client
- ${WANGLE}
- ${FIZZ}
- ${LIBSODIUM_LIBRARY}
- ${FOLLY_WITH_DEPENDENCIES}
- ${GLOG}
- ${GFLAGS_LIBRARIES}
-)
+target_link_libraries(cppDataSumWithReaderClient ${DATA_SUM_CLIENT_DEPS})
add_executable(cppDataSumWithWriterClient DataSumWithWriterClient.cpp)
+target_link_libraries(cppDataSumWithWriterClient ${DATA_SUM_CLIENT_DEPS})
-target_link_libraries(cppDataSumWithWriterClient dataSumWithWriterClient)
+add_executable(cppDataSumWithMergeWriterClient
DataSumWithMergeWriterClient.cpp)
+target_link_libraries(cppDataSumWithMergeWriterClient ${DATA_SUM_CLIENT_DEPS})
diff --git a/cpp/celeborn/tests/DataSumWithWriterClient.cpp
b/cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp
similarity index 87%
copy from cpp/celeborn/tests/DataSumWithWriterClient.cpp
copy to cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp
index ec7aeb5207..176139934c 100644
--- a/cpp/celeborn/tests/DataSumWithWriterClient.cpp
+++ b/cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp
@@ -16,7 +16,9 @@
*/
#include <folly/init/Init.h>
+#include <cassert>
#include <cstdio>
+#include <cstdlib>
#include <fstream>
#include <iostream>
@@ -24,7 +26,6 @@
int main(int argc, char** argv) {
folly::init(&argc, &argv, false);
- // Read the configs.
assert(argc == 10);
std::string lifecycleManagerHost = argv[1];
int lifecycleManagerPort = std::atoi(argv[2]);
@@ -44,7 +45,6 @@ int main(int argc, char** argv) {
<< ", resultFile = " << resultFile
<< ", compressCodec = " << compressCodec << std::endl;
- // Create shuffleClient and setup.
auto conf = std::make_shared<celeborn::conf::CelebornConf>();
conf->registerProperty(
celeborn::conf::CelebornConf::kShuffleCompressionCodec, compressCodec);
@@ -57,29 +57,28 @@ int main(int argc, char** argv) {
long maxData = 1000000;
size_t numData = 1000;
- // Generate data, sum up and pushData.
std::vector<long> result(numPartitions, 0);
std::vector<size_t> dataCnt(numPartitions, 0);
for (int mapId = 0; mapId < numMappers; mapId++) {
for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
- std::string partitionData;
for (size_t i = 0; i < numData; i++) {
int data = std::rand() % maxData;
result[partitionId] += data;
dataCnt[partitionId]++;
- partitionData += "-" + std::to_string(data);
+ std::string dataStr = "-" + std::to_string(data);
+ shuffleClient->mergeData(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ reinterpret_cast<const uint8_t*>(dataStr.c_str()),
+ 0,
+ dataStr.size(),
+ numMappers,
+ numPartitions);
}
- shuffleClient->pushData(
- shuffleId,
- mapId,
- attemptId,
- partitionId,
- reinterpret_cast<const uint8_t*>(partitionData.c_str()),
- 0,
- partitionData.size(),
- numMappers,
- numPartitions);
}
+ shuffleClient->pushMergedData(shuffleId, mapId, attemptId);
shuffleClient->mapperEnd(shuffleId, mapId, attemptId, numMappers);
}
for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
@@ -88,7 +87,6 @@ int main(int argc, char** argv) {
<< ", dataCnt = " << dataCnt[partitionId] << std::endl;
}
- // Write result to resultFile.
remove(resultFile.c_str());
std::ofstream of(resultFile);
for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
@@ -97,4 +95,4 @@ int main(int argc, char** argv) {
of.close();
return 0;
-}
+}
\ No newline at end of file
diff --git a/cpp/celeborn/tests/DataSumWithReaderClient.cpp
b/cpp/celeborn/tests/DataSumWithReaderClient.cpp
index 5333033230..5bcf677cc1 100644
--- a/cpp/celeborn/tests/DataSumWithReaderClient.cpp
+++ b/cpp/celeborn/tests/DataSumWithReaderClient.cpp
@@ -15,7 +15,9 @@
* limitations under the License.
*/
+#include <cassert>
#include <cstdio>
+#include <cstdlib>
#include <fstream>
#include <iostream>
diff --git a/cpp/celeborn/tests/DataSumWithWriterClient.cpp
b/cpp/celeborn/tests/DataSumWithWriterClient.cpp
index ec7aeb5207..ac1fcebf5b 100644
--- a/cpp/celeborn/tests/DataSumWithWriterClient.cpp
+++ b/cpp/celeborn/tests/DataSumWithWriterClient.cpp
@@ -16,7 +16,9 @@
*/
#include <folly/init/Init.h>
+#include <cassert>
#include <cstdio>
+#include <cstdlib>
#include <fstream>
#include <iostream>
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithLZ4.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithLZ4.scala
new file mode 100644
index 0000000000..ae7bf3c06f
--- /dev/null
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithLZ4.scala
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.cluster
+
+import org.apache.celeborn.common.protocol.CompressionCodec
+
+object CppMergeWriteJavaReadTestWithLZ4 extends JavaCppHybridReadWriteTestBase
{
+
+ def main(args: Array[String]) = {
+ testCppMergeWriteJavaRead(CompressionCodec.LZ4)
+ }
+}
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithNONE.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithNONE.scala
new file mode 100644
index 0000000000..8fa6df2879
--- /dev/null
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithNONE.scala
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.cluster
+
+import org.apache.celeborn.common.protocol.CompressionCodec
+
+object CppMergeWriteJavaReadTestWithNONE extends
JavaCppHybridReadWriteTestBase {
+
+ def main(args: Array[String]) = {
+ testCppMergeWriteJavaRead(CompressionCodec.NONE)
+ }
+}
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithZSTD.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithZSTD.scala
new file mode 100644
index 0000000000..91da455dde
--- /dev/null
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/CppMergeWriteJavaReadTestWithZSTD.scala
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.cluster
+
+import org.apache.celeborn.common.protocol.CompressionCodec
+
+object CppMergeWriteJavaReadTestWithZSTD extends
JavaCppHybridReadWriteTestBase {
+
+ def main(args: Array[String]) = {
+ testCppMergeWriteJavaRead(CompressionCodec.ZSTD)
+ }
+}
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala
index 621841826c..02e9792330 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/cluster/JavaCppHybridReadWriteTestBase.scala
@@ -18,7 +18,9 @@
package org.apache.celeborn.service.deploy.cluster
import java.io.File
+import java.nio.file.Files
import java.util
+import java.util.UUID
import scala.io.Source
import scala.util.Random
@@ -62,7 +64,7 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
}
def runJavaWriteCppRead(codec: CompressionCodec): Unit = {
- val appUniqueId = "test-app"
+ val appUniqueId = s"test-app-${UUID.randomUUID()}"
val shuffleId = 0
val attemptId = 0
@@ -81,71 +83,87 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
- // Generate random data, write with shuffleClient and calculate result.
- val numMappers = 2
- val numPartitions = 2
- val maxData = 1000000
- val numData = 1000
- var sums = new util.ArrayList[Long](numPartitions)
- val rand = new Random()
- var prefix = "-"
- if (codec != CompressionCodec.NONE) {
- // Add duplicate strings to make the compressed length shorter than the
original length.
- // Will be dropped in cpp test client.
- prefix = prefix ++ "++++++++++"
- }
- for (mapId <- 0 until numMappers) {
- for (partitionId <- 0 until numPartitions) {
- sums.add(0)
- for (i <- 0 until numData) {
- val data = rand.nextInt(maxData)
- sums.set(partitionId, sums.get(partitionId) + data)
- val dataStr = prefix + data.toString
- shuffleClient.pushOrMergeData(
- shuffleId,
- mapId,
- attemptId,
- partitionId,
- dataStr.getBytes,
- 0,
- dataStr.length,
- numMappers,
- numPartitions,
- false,
- false)
+ val cppResultFile = Files.createTempFile("celeborn-cpp-result-", ".txt")
+ try {
+ // Generate random data, write with shuffleClient and calculate result.
+ val numMappers = 2
+ val numPartitions = 2
+ val maxData = 1000000
+ val numData = 1000
+ var sums = new util.ArrayList[Long](numPartitions)
+ val rand = new Random()
+ var prefix = "-"
+ if (codec != CompressionCodec.NONE) {
+ // Add duplicate strings to make the compressed length shorter than
the original length.
+ // Will be dropped in cpp test client.
+ prefix = prefix ++ "++++++++++"
+ }
+ for (mapId <- 0 until numMappers) {
+ for (partitionId <- 0 until numPartitions) {
+ sums.add(0)
+ for (i <- 0 until numData) {
+ val data = rand.nextInt(maxData)
+ sums.set(partitionId, sums.get(partitionId) + data)
+ val dataStr = prefix + data.toString
+ shuffleClient.pushOrMergeData(
+ shuffleId,
+ mapId,
+ attemptId,
+ partitionId,
+ dataStr.getBytes,
+ 0,
+ dataStr.length,
+ numMappers,
+ numPartitions,
+ false,
+ false)
+ }
}
+ shuffleClient.pushMergedData(shuffleId, mapId, attemptId)
+ shuffleClient.mapperEnd(shuffleId, mapId, attemptId, numMappers,
numPartitions)
}
- shuffleClient.pushMergedData(shuffleId, mapId, attemptId)
- shuffleClient.mapperEnd(shuffleId, mapId, attemptId, numMappers,
numPartitions)
- }
- // Launch cpp reader to read data, calculate result and write to specific
result file.
- val cppResultFile = "/tmp/celeborn-cpp-result.txt"
- val lifecycleManagerHost = lifecycleManager.getHost
- val lifecycleManagerPort = lifecycleManager.getPort
- val projectDirectory = new File(new File(".").getAbsolutePath)
- val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
- val cppBinFileName = "cppDataSumWithReaderClient"
- val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
- val cppCodec = codec.name()
- // Execution command: $exec lifecycleManagerHost lifecycleManagerPort
appUniqueId shuffleId attemptId numPartitions cppResultFile cppCodec
- val command = {
- s"$cppBinFilePath $lifecycleManagerHost $lifecycleManagerPort
$appUniqueId $shuffleId $attemptId $numPartitions $cppResultFile $cppCodec"
- }
- println(s"run command: $command")
- val commandOutput = runCommand(command)
- println(s"command output: $commandOutput")
-
- // Verify the sum result.
- var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ // Launch cpp reader to read data, calculate result and write to
specific result file.
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithReaderClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ // Execution command: $exec lifecycleManagerHost lifecycleManagerPort
appUniqueId shuffleId attemptId numPartitions cppResultFile cppCodec
+ val command = Seq(
+ cppBinFilePath,
+ lifecycleManagerHost,
+ lifecycleManagerPort.toString,
+ appUniqueId,
+ shuffleId.toString,
+ attemptId.toString,
+ numPartitions.toString,
+ cppResultFile.toString,
+ cppCodec)
+ println(s"run command: ${command.mkString(" ")}")
+ val commandOutput = runCommand(command)
+ println(s"command output: $commandOutput")
+
+ // Verify the sum result.
+ var lineCount = 0
+ val source = Source.fromFile(cppResultFile.toFile, "utf-8")
+ try {
+ for (line <- source.getLines()) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ } finally {
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ Files.deleteIfExists(cppResultFile)
}
- Assert.assertEquals(lineCount, numPartitions)
- lifecycleManager.stop()
- shuffleClient.shutdown()
}
def testCppWriteJavaRead(codec: CompressionCodec): Unit = {
@@ -158,7 +176,7 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
}
def runCppWriteJavaRead(codec: CompressionCodec): Unit = {
- val appUniqueId = "test-app"
+ val appUniqueId = s"test-app-${UUID.randomUUID()}"
val shuffleId = 0
val attemptId = 0
@@ -177,71 +195,206 @@ trait JavaCppHybridReadWriteTestBase extends AnyFunSuite
new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
- val numMappers = 2
- val numPartitions = 2
-
- // Launch cpp writer to write data, calculate result and write to specific
result file.
- val cppResultFile = "/tmp/celeborn-cpp-writer-result.txt"
- val lifecycleManagerHost = lifecycleManager.getHost
- val lifecycleManagerPort = lifecycleManager.getPort
- val projectDirectory = new File(new File(".").getAbsolutePath)
- val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
- val cppBinFileName = "cppDataSumWithWriterClient"
- val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
- val cppCodec = codec.name()
- // Execution command: $exec lifecycleManagerHost lifecycleManagerPort
appUniqueId shuffleId attemptId numMappers numPartitions cppResultFile cppCodec
- val command = {
- s"$cppBinFilePath $lifecycleManagerHost $lifecycleManagerPort
$appUniqueId $shuffleId $attemptId $numMappers $numPartitions $cppResultFile
$cppCodec"
+ val cppResultFile = Files.createTempFile("celeborn-cpp-writer-result-",
".txt")
+ try {
+ val numMappers = 2
+ val numPartitions = 2
+
+ // Launch cpp writer to write data, calculate result and write to
specific result file.
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithWriterClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ // Execution command: $exec lifecycleManagerHost lifecycleManagerPort
appUniqueId shuffleId attemptId numMappers numPartitions cppResultFile cppCodec
+ val command = Seq(
+ cppBinFilePath,
+ lifecycleManagerHost,
+ lifecycleManagerPort.toString,
+ appUniqueId,
+ shuffleId.toString,
+ attemptId.toString,
+ numMappers.toString,
+ numPartitions.toString,
+ cppResultFile.toString,
+ cppCodec)
+ println(s"run command: ${command.mkString(" ")}")
+ val commandOutput = runCommand(command)
+ println(s"command output: $commandOutput")
+
+ val metricsCallback = new MetricsCallback {
+ override def incBytesRead(bytesWritten: Long): Unit = {}
+ override def incReadTime(time: Long): Unit = {}
+ }
+
+ var sums = new util.ArrayList[Long](numPartitions)
+ for (partitionId <- 0 until numPartitions) {
+ sums.add(0)
+ val inputStream = shuffleClient.readPartition(
+ shuffleId,
+ partitionId,
+ attemptId,
+ 0,
+ 0,
+ Integer.MAX_VALUE,
+ metricsCallback)
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
+ }
+ sums.set(partitionId, sums.get(partitionId) + data)
+ println(
+ s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
+ }
+ }
+
+ // Verify the sum result.
+ var lineCount = 0
+ val source = Source.fromFile(cppResultFile.toFile, "utf-8")
+ try {
+ for (line <- source.getLines()) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ } finally {
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ Files.deleteIfExists(cppResultFile)
}
- println(s"run command: $command")
- val commandOutput = runCommand(command)
- println(s"command output: $commandOutput")
+ }
- val metricsCallback = new MetricsCallback {
- override def incBytesRead(bytesWritten: Long): Unit = {}
- override def incReadTime(time: Long): Unit = {}
+ def testCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ beforeAll()
+ try {
+ runCppMergeWriteJavaRead(codec)
+ } finally {
+ afterAll()
}
+ }
+
+ def runCppMergeWriteJavaRead(codec: CompressionCodec): Unit = {
+ val appUniqueId = s"test-app-${UUID.randomUUID()}"
+ val shuffleId = 0
+ val attemptId = 0
- var sums = new util.ArrayList[Long](numPartitions)
- for (partitionId <- 0 until numPartitions) {
- sums.add(0)
- val inputStream = shuffleClient.readPartition(
- shuffleId,
- partitionId,
- attemptId,
- 0,
- 0,
- Integer.MAX_VALUE,
- metricsCallback)
- var c = inputStream.read()
- var data: Long = 0
- var dataCnt = 0
- while (c != -1) {
- if (c == '-') {
+ val clientConf = new CelebornConf()
+ .set(CelebornConf.MASTER_ENDPOINTS.key, s"localhost:$masterPort")
+ .set(CelebornConf.SHUFFLE_COMPRESSION_CODEC.key, codec.name)
+ .set(CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key, "true")
+ .set(CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key, "256K")
+ .set(CelebornConf.READ_LOCAL_SHUFFLE_FILE, false)
+ .set("celeborn.data.io.numConnectionsPerPeer", "1")
+ val lifecycleManager = new LifecycleManager(appUniqueId, clientConf)
+
+ val shuffleClient =
+ new ShuffleClientImpl(appUniqueId, clientConf, UserIdentifier("mock",
"mock"))
+ shuffleClient.setupLifecycleManagerRef(lifecycleManager.self)
+
+ val cppResultFile =
Files.createTempFile("celeborn-cpp-merge-writer-result-", ".txt")
+ try {
+ val numMappers = 2
+ val numPartitions = 2
+
+ val lifecycleManagerHost = lifecycleManager.getHost
+ val lifecycleManagerPort = lifecycleManager.getPort
+ val projectDirectory = new File(new File(".").getAbsolutePath)
+ val cppBinRelativeDirectory = "cpp/build/celeborn/tests/"
+ val cppBinFileName = "cppDataSumWithMergeWriterClient"
+ val cppBinFilePath =
s"$projectDirectory/$cppBinRelativeDirectory/$cppBinFileName"
+ val cppCodec = codec.name()
+ val command = Seq(
+ cppBinFilePath,
+ lifecycleManagerHost,
+ lifecycleManagerPort.toString,
+ appUniqueId,
+ shuffleId.toString,
+ attemptId.toString,
+ numMappers.toString,
+ numPartitions.toString,
+ cppResultFile.toString,
+ cppCodec)
+ println(s"run command: ${command.mkString(" ")}")
+ val commandOutput = runCommand(command)
+ println(s"command output: $commandOutput")
+
+ val metricsCallback = new MetricsCallback {
+ override def incBytesRead(bytesWritten: Long): Unit = {}
+ override def incReadTime(time: Long): Unit = {}
+ }
+
+ var sums = new util.ArrayList[Long](numPartitions)
+ for (partitionId <- 0 until numPartitions) {
+ sums.add(0)
+ val inputStream = shuffleClient.readPartition(
+ shuffleId,
+ partitionId,
+ attemptId,
+ 0,
+ 0,
+ Integer.MAX_VALUE,
+ metricsCallback)
+ try {
+ var c = inputStream.read()
+ var data: Long = 0
+ var dataCnt = 0
+ while (c != -1) {
+ if (c == '-') {
+ sums.set(partitionId, sums.get(partitionId) + data)
+ data = 0
+ dataCnt += 1
+ } else {
+ assert(c >= '0' && c <= '9')
+ data *= 10
+ data += c - '0'
+ }
+ c = inputStream.read()
+ }
sums.set(partitionId, sums.get(partitionId) + data)
- data = 0
- dataCnt += 1
- } else {
- assert(c >= '0' && c <= '9')
- data *= 10
- data += c - '0'
+ println(
+ s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
+ } finally {
+ inputStream.close()
}
- c = inputStream.read()
}
- sums.set(partitionId, sums.get(partitionId) + data)
- println(s"partition $partitionId sum result = ${sums.get(partitionId)},
dataCnt = $dataCnt")
- }
- // Verify the sum result.
- var lineCount = 0
- for (line <- Source.fromFile(cppResultFile, "utf-8").getLines.toList) {
- val data = line.toLong
- Assert.assertEquals(data, sums.get(lineCount))
- lineCount += 1
+ var lineCount = 0
+ val source = Source.fromFile(cppResultFile.toFile, "utf-8")
+ try {
+ for (line <- source.getLines()) {
+ val data = line.toLong
+ Assert.assertEquals(data, sums.get(lineCount))
+ lineCount += 1
+ }
+ } finally {
+ source.close()
+ }
+ Assert.assertEquals(lineCount, numPartitions)
+ } finally {
+ lifecycleManager.stop()
+ shuffleClient.shutdown()
+ Files.deleteIfExists(cppResultFile)
}
- Assert.assertEquals(lineCount, numPartitions)
- lifecycleManager.stop()
- shuffleClient.shutdown()
}
}
