From bd9e0ddc1f97db2ef5bad8ddfa6fde5e61a90a3a Mon Sep 17 00:00:00 2001 From: jiaoqingbo <1178404354@qq.com> Date: Wed, 15 Feb 2023 15:49:08 +0800 Subject: [PATCH] [CELEBORN-304] Missing setIfMissing `celeborn.$module.io.serverThreads` (#1238) --- .../sql/execution/columnar/RssCompressibleColumnBuilder.scala | 2 +- .../main/java/org/apache/celeborn/client/ShuffleClient.java | 4 ++-- .../main/scala/org/apache/celeborn/common/util/Utils.scala | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/client-spark/spark-3/src/main/scala/org/apache/spark/sql/execution/columnar/RssCompressibleColumnBuilder.scala b/client-spark/spark-3/src/main/scala/org/apache/spark/sql/execution/columnar/RssCompressibleColumnBuilder.scala index 9188b0a38..456037082 100644 --- a/client-spark/spark-3/src/main/scala/org/apache/spark/sql/execution/columnar/RssCompressibleColumnBuilder.scala +++ b/client-spark/spark-3/src/main/scala/org/apache/spark/sql/execution/columnar/RssCompressibleColumnBuilder.scala @@ -44,7 +44,7 @@ trait RssCompressibleColumnBuilder[T <: AtomicType] // The various compression schemes, while saving memory use, cause all of the data within // the row to become unaligned, thus causing crashes. Until a way of fixing the compression - // is found to also allow aligned accesses this must be disabled for SPARC. + // is found to also allow aligned accesses this must be disabled for SPARK. protected def isWorthCompressing(encoder: Encoder[T]) = { RssCompressibleColumnBuilder.unaligned && encoder.compressionRatio < 0.8 diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java index c8c5b4ab4..184400b30 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java @@ -199,7 +199,7 @@ public abstract class ShuffleClient { // Write data to a specific map partition, input data's type is Bytebuf. // data's type is Bytebuf to avoid copy between application and netty - // closecallback will do some clean opertions like memory release. + // closecallback will do some clean operations like memory release. public abstract int pushDataToLocation( String applicationId, int shuffleId, @@ -209,7 +209,7 @@ public abstract class ShuffleClient { ByteBuf data, PartitionLocation location, BooleanSupplier closeCallBack) - throws IOException;; + throws IOException; public abstract Optional regionStart( String applicationId, 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 0926ac88f..3559c13c0 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 @@ -497,7 +497,7 @@ object Utils extends Logging { // assuming we have all the machine's cores). // NB: Only set if serverThreads/clientThreads not already set. val numThreads = defaultNumThreads(numUsableCores) - conf.setIfMissing(s"celeborn.$module.io.clientThreads", numThreads.toString) + conf.setIfMissing(s"celeborn.$module.io.serverThreads", numThreads.toString) conf.setIfMissing(s"celeborn.$module.io.clientThreads", numThreads.toString) // TODO remove after releasing 0.2.0 conf.setIfMissing(s"rss.$module.io.serverThreads", numThreads.toString)