Merge branch 'master' into issue-1749

bin/start-coordinator.sh

@@ -30,6 +30,7 @@ COORDINATOR_CONF_FILE="${RSS_CONF_DIR}/coordinator.conf"
 JAR_DIR="${RSS_HOME}/jars"
 LOG_CONF_FILE="${RSS_CONF_DIR}/log4j2.xml"
 LOG_PATH="${RSS_LOG_DIR}/coordinator.log"
+LOG_OUT_PATH="${RSS_LOG_DIR}/coordinator.out"
 COORDINATOR_RPC_AUDIT_LOG_PATH="${RSS_LOG_DIR}/coordinator_rpc_audit.log"
 
 MAIN_CLASS="org.apache.uniffle.coordinator.CoordinatorServer"
@@ -97,7 +98,7 @@ GC_LOG_ARGS_NEW=" -XX:+IgnoreUnrecognizedVMOptions \
 JVM_LOG_ARGS=""
 
 if [ -f ${LOG_CONF_FILE} ]; then
- JVM_LOG_ARGS=" -Dlog4j2.configurationFile=file:${LOG_CONF_FILE} -Dlog.path=${LOG_PATH} -Dcoordinator.rpc.audit.log.path=${COORDINATOR_RPC_AUDIT_LOG_PATH}"
+ JVM_LOG_ARGS=" -Dlog4j2.configurationFile=file:${LOG_CONF_FILE} -Dlog.path=${LOG_PATH} -Drpc.audit.log.path=${COORDINATOR_RPC_AUDIT_LOG_PATH}"
 else
  echo "Exit with error: ${LOG_CONF_FILE} file doesn't exist."
  exit 1
@@ -111,7 +112,7 @@ else
 fi
 
 COORDINATOR_JAVA_OPTS=${COORDINATOR_JAVA_OPTS:-""}
-$RUNNER ${COORDINATOR_BASE_JVM_ARGS} ${COORDINATOR_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${COORDINATOR_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${COORDINATOR_CONF_FILE}" $@ &
+(nohup $RUNNER ${COORDINATOR_BASE_JVM_ARGS} ${COORDINATOR_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${COORDINATOR_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${COORDINATOR_CONF_FILE}" $@ > ${LOG_OUT_PATH} 2>&1) &
 
 get_pid_file_name coordinator
 echo $! >${RSS_PID_DIR}/${pid_file}

bin/start-dashboard.sh

@@ -30,6 +30,7 @@ DASHBOARD_CONF_FILE="${RSS_CONF_DIR}/dashboard.conf"
 JAR_DIR="${RSS_HOME}/jars"
 LOG_CONF_FILE="${RSS_CONF_DIR}/log4j2.xml"
 LOG_PATH="${RSS_LOG_DIR}/dashboard.log"
+LOG_OUT_PATH="${RSS_LOG_DIR}/dashboard.out"
 
 MAIN_CLASS="org.apache.uniffle.dashboard.web.Dashboard"
 
@@ -91,7 +92,6 @@ else
 fi
 
 DASHBOARD_JAVA_OPTS=${DASHBOARD_JAVA_OPTS:-""}
-$RUNNER ${DASHBOARD_BASE_JVM_ARGS} ${DASHBOARD_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${DASHBOARD_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${DASHBOARD_CONF_FILE}" $@ &
-
+(nohup $RUNNER ${DASHBOARD_BASE_JVM_ARGS} ${DASHBOARD_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${DASHBOARD_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${DASHBOARD_CONF_FILE}" $@ > ${LOG_OUT_PATH} 2>&1) &
 get_pid_file_name dashboard
 echo $! >${RSS_PID_DIR}/${pid_file}

bin/start-shuffle-server.sh

@@ -30,6 +30,7 @@ SHUFFLE_SERVER_CONF_FILE="${RSS_CONF_DIR}/server.conf"
 JAR_DIR="${RSS_HOME}/jars"
 LOG_CONF_FILE="${RSS_CONF_DIR}/log4j2.xml"
 LOG_PATH="${RSS_LOG_DIR}/shuffle_server.log"
+LOG_OUT_PATH="${RSS_LOG_DIR}/shuffle_server.out"
 SHUFFLE_SERVER_STORAGE_AUDIT_LOG_PATH=${SHUFFLE_SERVER_STORAGE_AUDIT_LOG_PATH:-"${RSS_LOG_DIR}/shuffle_server_storage_audit.log"}
 SHUFFLE_SERVER_RPC_AUDIT_LOG_PATH=${SHUFFLE_SERVER_RPC_AUDIT_LOG_PATH:-"${RSS_LOG_DIR}/shuffle_server_rpc_audit.log"}
 
@@ -135,7 +136,7 @@ GC_LOG_ARGS_NEW=" -XX:+IgnoreUnrecognizedVMOptions \
 JVM_LOG_ARGS=""
 
 if [ -f ${LOG_CONF_FILE} ]; then
- JVM_LOG_ARGS=" -Dlog4j2.configurationFile=file:${LOG_CONF_FILE} -Dlog.path=${LOG_PATH} -Dshuffle.server.storage.audit.log.path=${SHUFFLE_SERVER_STORAGE_AUDIT_LOG_PATH} -Dshuffle.server.rpc.audit.log.path=${SHUFFLE_SERVER_RPC_AUDIT_LOG_PATH}"
+ JVM_LOG_ARGS=" -Dlog4j2.configurationFile=file:${LOG_CONF_FILE} -Dlog.path=${LOG_PATH} -Dstorage.audit.log.path=${SHUFFLE_SERVER_STORAGE_AUDIT_LOG_PATH} -Drpc.audit.log.path=${SHUFFLE_SERVER_RPC_AUDIT_LOG_PATH}"
 else
  echo "Exit with error: ${LOG_CONF_FILE} file doesn't exist."
  exit 1
@@ -149,7 +150,7 @@ else
 fi
 
 SHUFFLE_SERVER_JAVA_OPTS=${SHUFFLE_SERVER_JAVA_OPTS:-""}
-$RUNNER ${SHUFFLE_SERVER_BASE_JVM_ARGS} ${SHUFFLE_SERVER_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${JAVA_LIB_PATH} ${SHUFFLE_SERVER_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${SHUFFLE_SERVER_CONF_FILE}" $@ &
+(nohup $RUNNER ${SHUFFLE_SERVER_BASE_JVM_ARGS} ${SHUFFLE_SERVER_JVM_GC_ARGS} ${JVM_LOG_ARGS} ${JAVA_LIB_PATH} ${SHUFFLE_SERVER_JAVA_OPTS} -cp ${CLASSPATH} ${MAIN_CLASS} --conf "${SHUFFLE_SERVER_CONF_FILE}" $@ > ${LOG_OUT_PATH} 2>&1) &
 
 get_pid_file_name shuffle-server
 echo $! >${RSS_PID_DIR}/${pid_file}

client-mr/core/src/main/java/org/apache/hadoop/mapred/SortWriteBufferManager.java

@@ -23,6 +23,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -96,7 +97,7 @@ public class SortWriteBufferManager<K, V> {
  private final long maxBufferSize;
  private final ExecutorService sendExecutorService;
  private final RssConf rssConf;
- private final Codec codec;
+ private final Optional<Codec> codec;
  private final Task.CombinerRunner<K, V> combinerRunner;
 
  public SortWriteBufferManager(
@@ -383,7 +384,7 @@ ShuffleBlockInfo createShuffleBlock(SortWriteBuffer wb) {
  int partitionId = wb.getPartitionId();
  final int uncompressLength = data.length;
  long start = System.currentTimeMillis();
- final byte[] compressed = codec.compress(data);
+ final byte[] compressed = codec.map(c -> c.compress(data)).orElse(data);
  final long crc32 = ChecksumUtils.getCrc32(compressed);
  compressTime += System.currentTimeMillis() - start;
  final long blockId =

client-mr/core/src/main/java/org/apache/hadoop/mapreduce/task/reduce/RssFetcher.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.text.DecimalFormat;
+import java.util.Optional;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.mapred.Counters;
@@ -38,6 +39,7 @@
 import org.apache.uniffle.common.compression.Codec;
 import org.apache.uniffle.common.config.RssConf;
 import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.common.util.ByteBufferUtils;
 import org.apache.uniffle.common.util.ByteUnit;
 
 public class RssFetcher<K, V> {
@@ -90,7 +92,7 @@ private enum ShuffleErrors {
  private int waitCount = 0;
  private byte[] uncompressedData = null;
  private RssConf rssConf;
- private Codec codec;
+ private Optional<Codec> codec;
 
  RssFetcher(
  JobConf job,
@@ -161,14 +163,19 @@ public void copyFromRssServer() throws IOException {
 
  // uncompress the block
  if (!hasPendingData && compressedData != null) {
- final long startDecompress = System.currentTimeMillis();
- int uncompressedLen = compressedBlock.getUncompressLength();
- ByteBuffer decompressedBuffer = ByteBuffer.allocate(uncompressedLen);
- codec.decompress(compressedData, uncompressedLen, decompressedBuffer, 0);
- uncompressedData = decompressedBuffer.array();
- unCompressionLength += compressedBlock.getUncompressLength();
- long decompressDuration = System.currentTimeMillis() - startDecompress;
- decompressTime += decompressDuration;
+ if (codec.isPresent()) {
+ final long startDecompress = System.currentTimeMillis();
+ int uncompressedLen = compressedBlock.getUncompressLength();
+ ByteBuffer decompressedBuffer = ByteBuffer.allocate(uncompressedLen);
+ codec.get().decompress(compressedData, uncompressedLen, decompressedBuffer, 0);
+ uncompressedData = decompressedBuffer.array();
+ unCompressionLength += compressedBlock.getUncompressLength();
+ long decompressDuration = System.currentTimeMillis() - startDecompress;
+ decompressTime += decompressDuration;
+ } else {
+ uncompressedData = ByteBufferUtils.bufferToArray(compressedData);
+ unCompressionLength += uncompressedData.length;
+ }
  }
 
  if (uncompressedData != null) {

client-spark/common/src/main/java/org/apache/spark/shuffle/reader/RssShuffleDataIterator.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Optional;
 
 import scala.Product2;
 import scala.Tuple2;
@@ -59,7 +60,7 @@ public class RssShuffleDataIterator<K, C> extends AbstractIterator<Product2<K, C
  private long totalRawBytesLength = 0;
  private long unCompressedBytesLength = 0;
  private ByteBuffer uncompressedData;
- private Codec codec;
+ private Optional<Codec> codec;
 
  public RssShuffleDataIterator(
  Serializer serializer,
@@ -74,7 +75,7 @@ public RssShuffleDataIterator(
  RssSparkConfig.SPARK_SHUFFLE_COMPRESS_KEY.substring(
  RssSparkConfig.SPARK_RSS_CONFIG_PREFIX.length()),
  RssSparkConfig.SPARK_SHUFFLE_COMPRESS_DEFAULT);
- this.codec = compress ? Codec.newInstance(rssConf) : null;
+ this.codec = compress ? Codec.newInstance(rssConf) : Optional.empty();
  }
 
  public Iterator<Tuple2<Object, Object>> createKVIterator(ByteBuffer data) {
@@ -131,7 +132,7 @@ public boolean hasNext() {
  shuffleReadClient.checkProcessedBlockIds();
  shuffleReadClient.logStatics();
  String decInfo =
- codec == null
+ !codec.isPresent()
  ? "."
  : (", "
  + decompressTime
@@ -160,7 +161,7 @@ private int uncompress(CompressedShuffleBlock rawBlock, ByteBuffer rawData) {
  shuffleReadMetrics.incRemoteBytesRead(rawDataLength);
 
  int uncompressedLen = rawBlock.getUncompressLength();
- if (codec != null) {
+ if (codec.isPresent()) {
  if (uncompressedData == null
  || uncompressedData.capacity() < uncompressedLen
  || !isSameMemoryType(uncompressedData, rawData)) {
@@ -185,7 +186,7 @@ private int uncompress(CompressedShuffleBlock rawBlock, ByteBuffer rawData) {
  }
  uncompressedData.clear();
  long startDecompress = System.currentTimeMillis();
- codec.decompress(rawData, uncompressedLen, uncompressedData, 0);
+ codec.get().decompress(rawData, uncompressedLen, uncompressedData, 0);
  unCompressedBytesLength += uncompressedLen;
  long decompressDuration = System.currentTimeMillis() - startDecompress;
  decompressTime += decompressDuration;
@@ -210,7 +211,7 @@ public BoxedUnit cleanup() {
  // Uncompressed data is released in this class, Compressed data is release in the class
  // ShuffleReadClientImpl
  // So if codec is null, we don't release the data when the stream is closed
- if (codec != null) {
+ if (codec.isPresent()) {
  RssUtils.releaseByteBuffer(uncompressedData);
  }
  if (shuffleReadClient != null) {

client-spark/common/src/main/java/org/apache/spark/shuffle/writer/WriteBufferManager.java

@@ -22,6 +22,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -91,7 +92,7 @@ public class WriteBufferManager extends MemoryConsumer {
  private long uncompressedDataLen = 0;
  private long requireMemoryInterval;
  private int requireMemoryRetryMax;
- private Codec codec;
+ private Optional<Codec> codec;
  private Function<List<ShuffleBlockInfo>, List<CompletableFuture<Long>>> spillFunc;
  private long sendSizeLimit;
  private boolean memorySpillEnabled;
@@ -159,7 +160,7 @@ public WriteBufferManager(
  RssSparkConfig.SPARK_SHUFFLE_COMPRESS_KEY.substring(
  RssSparkConfig.SPARK_RSS_CONFIG_PREFIX.length()),
  RssSparkConfig.SPARK_SHUFFLE_COMPRESS_DEFAULT);
- this.codec = compress ? Codec.newInstance(rssConf) : null;
+ this.codec = compress ? Codec.newInstance(rssConf) : Optional.empty();
  this.spillFunc = spillFunc;
  this.sendSizeLimit = rssConf.get(RssSparkConfig.RSS_CLIENT_SEND_SIZE_LIMITATION);
  this.memorySpillTimeoutSec = rssConf.get(RssSparkConfig.RSS_MEMORY_SPILL_TIMEOUT);
@@ -384,9 +385,9 @@ protected ShuffleBlockInfo createShuffleBlock(int partitionId, WriterBuffer wb)
  byte[] data = wb.getData();
  final int uncompressLength = data.length;
  byte[] compressed = data;
- if (codec != null) {
+ if (codec.isPresent()) {
  long start = System.currentTimeMillis();
- compressed = codec.compress(data);
+ compressed = codec.get().compress(data);
  compressTime += System.currentTimeMillis() - start;
  }
  final long crc32 = ChecksumUtils.getCrc32(compressed);

client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/RssShuffleManagerBase.java

@@ -37,7 +37,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.spark.MapOutputTracker;

client-spark/common/src/test/java/org/apache/spark/shuffle/reader/AbstractRssReaderTest.java

@@ -19,6 +19,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -171,8 +172,9 @@ protected ShufflePartitionedBlock createShuffleBlock(byte[] data, long blockId)
  protected ShufflePartitionedBlock createShuffleBlock(
  byte[] data, long blockId, boolean compress) {
  byte[] compressData = data;
- if (compress) {
- compressData = Codec.newInstance(new RssConf()).compress(data);
+ Optional<Codec> codec = Codec.newInstance(new RssConf());
+ if (compress && codec.isPresent()) {
+ compressData = codec.get().compress(data);
  }
  long crc = ChecksumUtils.getCrc32(compressData);
  return new ShufflePartitionedBlock(

client-spark/common/src/test/java/org/apache/spark/shuffle/reader/RssShuffleDataIteratorTest.java

@@ -20,6 +20,7 @@
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Stream;
 
 import com.google.common.collect.Lists;
@@ -46,6 +47,7 @@
 import org.apache.uniffle.client.impl.ShuffleReadClientImpl;
 import org.apache.uniffle.common.ClientType;
 import org.apache.uniffle.common.ShuffleServerInfo;
+import org.apache.uniffle.common.compression.Codec;
 import org.apache.uniffle.common.config.RssConf;
 import org.apache.uniffle.common.util.BlockIdLayout;
 import org.apache.uniffle.common.util.ChecksumUtils;
@@ -321,11 +323,12 @@ private void readTestCompressOrNot(String path, boolean compress) throws Excepti
  RssShuffleDataIterator rssShuffleDataIterator =
  getDataIterator(
  basePath, blockIdBitmap, taskIdBitmap, Lists.newArrayList(ssi1, ssi2), compress);
- Object codec = FieldUtils.readField(rssShuffleDataIterator, "codec", true);
+ Optional<Codec> codec =
+ (Optional<Codec>) FieldUtils.readField(rssShuffleDataIterator, "codec", true);
  if (compress) {
- Assertions.assertNotNull(codec);
+ Assertions.assertTrue(codec.isPresent());
  } else {
- Assertions.assertNull(codec);
+ Assertions.assertFalse(codec.isPresent());
  }
 
  validateResult(rssShuffleDataIterator, expectedData, 20);

client-spark/common/src/test/java/org/apache/spark/shuffle/writer/WriteBufferManagerTest.java

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -46,6 +47,7 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.uniffle.common.ShuffleBlockInfo;
+import org.apache.uniffle.common.compression.Codec;
 import org.apache.uniffle.common.config.RssClientConf;
 import org.apache.uniffle.common.config.RssConf;
 import org.apache.uniffle.common.util.BlockIdLayout;
@@ -122,11 +124,11 @@ private void addRecord(boolean compress, BlockIdLayout layout) throws IllegalAcc
  conf.set(RssSparkConfig.SPARK_SHUFFLE_COMPRESS_KEY, String.valueOf(false));
  }
  WriteBufferManager wbm = createManager(conf);
- Object codec = FieldUtils.readField(wbm, "codec", true);
+ Optional<Codec> codec = (Optional<Codec>) FieldUtils.readField(wbm, "codec", true);
  if (compress) {
- Assertions.assertNotNull(codec);
+ Assertions.assertTrue(codec.isPresent());
  } else {
- Assertions.assertNull(codec);
+ Assertions.assertFalse(codec.isPresent());
  }
  wbm.setShuffleWriteMetrics(new ShuffleWriteMetrics());
  String testKey = "Key";

client-spark/spark2-shaded/pom.xml

@@ -312,6 +312,9 @@
  <echo message="repackaging netty jar"></echo>
  <jar destfile="${project.build.directory}/${project.artifactId}-${project.version}.jar"
  basedir="${project.build.directory}/unpacked"/>
+ <exec executable="bash" failonerror="true" dir="${project.build.directory}/">
+ <arg line="${project.basedir}/../../dev/scripts/checkshade.sh ${project.build.directory}/${project.artifactId}-${project.version}.jar"/>
+ </exec>
  </target>
  </configuration>
  </execution>

client-spark/spark3-shaded/pom.xml

@@ -309,6 +309,9 @@
  <echo message="repackaging netty jar"></echo>
  <jar destfile="${project.build.directory}/${project.artifactId}-${project.version}.jar"
  basedir="${project.build.directory}/unpacked"/>
+ <exec executable="bash" failonerror="true" dir="${project.build.directory}/">
+ <arg line="${project.basedir}/../../dev/scripts/checkshade.sh ${project.build.directory}/${project.artifactId}-${project.version}.jar"/>
+ </exec>
  </target>
  </configuration>
  </execution>