[CELEBORN-2275][CIP-14] Add C++ merge-write and Java-read hybrid integration test

.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"
+            -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"

cpp/celeborn/tests/CMakeLists.txt

@@ -61,3 +61,28 @@ target_link_libraries(
 add_executable(cppDataSumWithWriterClient DataSumWithWriterClient.cpp)
 
 target_link_libraries(cppDataSumWithWriterClient dataSumWithWriterClient)
+
+add_library(
+        dataSumWithMergeWriterClient
+        DataSumWithMergeWriterClient.cpp)
+
+target_link_libraries(
+        dataSumWithMergeWriterClient
+        memory
+        utils
+        conf
+        proto
+        network
+        protocol
+        client
+        ${WANGLE}
+        ${FIZZ}
+        ${LIBSODIUM_LIBRARY}
+        ${FOLLY_WITH_DEPENDENCIES}
+        ${GLOG}
+        ${GFLAGS_LIBRARIES}
+)
+
+add_executable(cppDataSumWithMergeWriterClient DataSumWithMergeWriterClient.cpp)
+
+target_link_libraries(cppDataSumWithMergeWriterClient dataSumWithMergeWriterClient)

cpp/celeborn/tests/DataSumWithMergeWriterClient.cpp

@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+#include <folly/init/Init.h>
+#include <cstdio>
+#include <fstream>
+#include <iostream>
+
+#include <celeborn/client/ShuffleClient.h>
+
+int main(int argc, char** argv) {
+  folly::init(&argc, &argv, false);
+  assert(argc == 10);
+  std::string lifecycleManagerHost = argv[1];
+  int lifecycleManagerPort = std::atoi(argv[2]);
+  std::string appUniqueId = argv[3];
+  int shuffleId = std::atoi(argv[4]);
+  int attemptId = std::atoi(argv[5]);
+  int numMappers = std::atoi(argv[6]);
+  int numPartitions = std::atoi(argv[7]);
+  std::string resultFile = argv[8];
+  std::string compressCodec = argv[9];
+  std::cout << "lifecycleManagerHost = " << lifecycleManagerHost
+            << ", lifecycleManagerPort = " << lifecycleManagerPort
+            << ", appUniqueId = " << appUniqueId
+            << ", shuffleId = " << shuffleId << ", attemptId = " << attemptId
+            << ", numMappers = " << numMappers
+            << ", numPartitions = " << numPartitions
+            << ", resultFile = " << resultFile
+            << ", compressCodec = " << compressCodec << std::endl;
+
+  auto conf = std::make_shared<celeborn::conf::CelebornConf>();
+  conf->registerProperty(
+      celeborn::conf::CelebornConf::kShuffleCompressionCodec, compressCodec);
+  auto clientEndpoint =
+      std::make_shared<celeborn::client::ShuffleClientEndpoint>(conf);
+  auto shuffleClient = celeborn::client::ShuffleClientImpl::create(
+      appUniqueId, conf, *clientEndpoint);
+  shuffleClient->setupLifecycleManagerRef(
+      lifecycleManagerHost, lifecycleManagerPort);
+
+  long maxData = 1000000;
+  size_t numData = 1000;
+  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++) {
+      for (size_t i = 0; i < numData; i++) {
+        int data = std::rand() % maxData;
+        result[partitionId] += data;
+        dataCnt[partitionId]++;
+        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->pushMergedData(shuffleId, mapId, attemptId);
+    shuffleClient->mapperEnd(shuffleId, mapId, attemptId, numMappers);
+  }
+  for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+    std::cout << "partition " << partitionId
+              << " sum result = " << result[partitionId]
+              << ", dataCnt = " << dataCnt[partitionId] << std::endl;
+  }
+
+  remove(resultFile.c_str());
+  std::ofstream of(resultFile);
+  for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+    of << result[partitionId] << std::endl;
+  }
+  of.close();
+
+  return 0;
+}

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)
+  }
+}

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)
+  }
+}

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)
+  }
+}