This is an automated email from the ASF dual-hosted git repository.
nicholasjiang 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 954277d07 [CELEBORN-1254] PartitionFilesSorter seeks to position of
each block and does not warm up for non-hdfs files
954277d07 is described below
commit 954277d07c7a87731f888058034036c9aebd84a9
Author: SteNicholas <[email protected]>
AuthorDate: Fri Jan 26 19:20:06 2024 +0800
[CELEBORN-1254] PartitionFilesSorter seeks to position of each block and
does not warm up for non-hdfs files
### What changes were proposed in this pull request?
Introduce `celeborn.worker.sortPartition.reservedMemory.enabled` to support
that `PartitionFilesSorter` seeks to position of each block and does not warmup
for non-hdfs files.
### Why are the changes needed?
File sorting includes three steps: reading files, sorting MapIds, and
writing files. The default block of Celeborn is 256k, and the number of blocks
is about 1000, so the sorting process is very fast, and the main overhead is
file reading and writing. There are roughly three options for the entire
sorting process:
1. Memory of the file size is allocated in advance, the file is read in as
a whole, MapId is parsed and sorted, and Blocks are written back to the disk in
MapId order.
2. No memory is allocated, seek to the location of each block, parse and
sort the MapId, and transfer the Blocks of the original file to the new file in
the order of MapId.
3. Allocate a small block of memory (such as 256k), read the entire file
sequentially, parse and sort the MapId, and transfer the block of the original
file to the new file in the order of MapId.
From an IO perspective, at first glance, solution 1 uses sufficient memory
and there is no sequential reading and writing; solution 2 has random reading
and random writing; solution 3 has sequential writing. Intuitively solution 1
has better performance. Due to the existence of PageCache, when writing a file
in solution 3, the original file is likely to be cached in PageCache.
`PartitionFilesSorter` support solution3 with PageCache at present, which has
better performance especially H [...]
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
GA and cluster.
Closes #2264 from SteNicholas/CELEBORN-1254.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
.../org/apache/celeborn/common/CelebornConf.scala | 43 ++++++++++++++--------
docs/configuration/worker.md | 1 +
.../deploy/worker/memory/MemoryManager.java | 2 +-
.../worker/storage/PartitionFilesSorter.java | 40 ++++++++++++--------
.../worker/storage/CreditStreamManagerSuiteJ.java | 2 +-
.../storage/MapPartitionDataWriterSuiteJ.java | 2 +-
.../worker/storage/PartitionFilesSorterSuiteJ.java | 2 +-
.../storage/ReducePartitionDataWriterSuiteJ.java | 2 +-
8 files changed, 59 insertions(+), 35 deletions(-)
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index ca2f8f132..887be0830 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -683,13 +683,16 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
def workerCleanThreads: Int = get(WORKER_CLEAN_THREADS)
def workerShuffleCommitTimeout: Long = get(WORKER_SHUFFLE_COMMIT_TIMEOUT)
def minPartitionSizeToEstimate: Long = get(ESTIMATED_PARTITION_SIZE_MIN_SIZE)
- def partitionSorterSortPartitionTimeout: Long =
get(PARTITION_SORTER_SORT_TIMEOUT)
- def partitionSorterReservedMemoryPerPartition: Long =
- get(WORKER_PARTITION_SORTER_PER_PARTITION_RESERVED_MEMORY)
- def partitionSorterThreads: Int =
-
get(PARTITION_SORTER_THREADS).getOrElse(Runtime.getRuntime.availableProcessors)
- def partitionSorterIndexCacheMaxWeight: Long =
get(PARTITION_SORTER_INDEX_CACHE_MAX_WEIGHT)
- def partitionSorterIndexExpire: Long =
get(PARTITION_SORTER_INDEX_CACHE_EXPIRE)
+ def workerPartitionSorterSortPartitionTimeout: Long =
get(WORKER_PARTITION_SORTER_SORT_TIMEOUT)
+ def workerPartitionSorterReservedMemoryEnabled: Boolean =
+ get(WORKER_PARTITION_SORTER_RESERVED_MEMORY_ENABLED)
+ def workerPartitionSorterReservedMemoryPerPartition: Long =
+ get(WORKER_PARTITION_SORTER_RESERVED_MEMORY_PER_PARTITION)
+ def workerPartitionSorterThreads: Int =
+
get(WORKER_PARTITION_SORTER_THREADS).getOrElse(Runtime.getRuntime.availableProcessors)
+ def workerPartitionSorterIndexCacheMaxWeight: Long =
+ get(WORKER_PARTITION_SORTER_INDEX_CACHE_MAX_WEIGHT)
+ def workerPartitionSorterIndexExpire: Long =
get(WORKER_PARTITION_SORTER_INDEX_CACHE_EXPIRE)
def workerPushHeartbeatEnabled: Boolean = get(WORKER_PUSH_HEARTBEAT_ENABLED)
def workerPushMaxComponents: Int =
get(WORKER_PUSH_COMPOSITEBUFFER_MAXCOMPONENTS)
def workerFetchHeartbeatEnabled: Boolean =
get(WORKER_FETCH_HEARTBEAT_ENABLED)
@@ -1042,8 +1045,8 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
def workerDirectMemoryRatioToPauseReplicate: Double =
get(WORKER_DIRECT_MEMORY_RATIO_PAUSE_REPLICATE)
def workerDirectMemoryRatioToResume: Double =
get(WORKER_DIRECT_MEMORY_RATIO_RESUME)
- def partitionSorterDirectMemoryRatioThreshold: Double =
- get(PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD)
+ def workerPartitionSorterDirectMemoryRatioThreshold: Double =
+ get(WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD)
def workerDirectMemoryPressureCheckIntervalMs: Long =
get(WORKER_DIRECT_MEMORY_CHECK_INTERVAL)
def workerDirectMemoryReportIntervalSecond: Long =
get(WORKER_DIRECT_MEMORY_REPORT_INTERVAL)
def workerDirectMemoryTrimChannelWaitInterval: Long =
@@ -2486,7 +2489,7 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("120s")
- val PARTITION_SORTER_SORT_TIMEOUT: ConfigEntry[Long] =
+ val WORKER_PARTITION_SORTER_SORT_TIMEOUT: ConfigEntry[Long] =
buildConf("celeborn.worker.sortPartition.timeout")
.withAlternative("celeborn.worker.partitionSorter.sort.timeout")
.categories("worker")
@@ -2495,7 +2498,7 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("220s")
- val PARTITION_SORTER_THREADS: OptionalConfigEntry[Int] =
+ val WORKER_PARTITION_SORTER_THREADS: OptionalConfigEntry[Int] =
buildConf("celeborn.worker.sortPartition.threads")
.withAlternative("celeborn.worker.partitionSorter.threads")
.categories("worker")
@@ -2505,7 +2508,7 @@ object CelebornConf extends Logging {
.intConf
.createOptional
- val PARTITION_SORTER_INDEX_CACHE_MAX_WEIGHT: ConfigEntry[Long] =
+ val WORKER_PARTITION_SORTER_INDEX_CACHE_MAX_WEIGHT: ConfigEntry[Long] =
buildConf("celeborn.worker.sortPartition.indexCache.maxWeight")
.categories("worker")
.doc("PartitionSorter's cache max weight for index buffer.")
@@ -2513,7 +2516,7 @@ object CelebornConf extends Logging {
.longConf
.createWithDefault(100000)
- val PARTITION_SORTER_INDEX_CACHE_EXPIRE: ConfigEntry[Long] =
+ val WORKER_PARTITION_SORTER_INDEX_CACHE_EXPIRE: ConfigEntry[Long] =
buildConf("celeborn.worker.sortPartition.indexCache.expire")
.categories("worker")
.doc("PartitionSorter's cache item expire time.")
@@ -2521,7 +2524,7 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("180s")
- val WORKER_PARTITION_SORTER_PER_PARTITION_RESERVED_MEMORY: ConfigEntry[Long]
=
+ val WORKER_PARTITION_SORTER_RESERVED_MEMORY_PER_PARTITION: ConfigEntry[Long]
=
buildConf("celeborn.worker.sortPartition.reservedMemoryPerPartition")
.withAlternative("celeborn.worker.partitionSorter.reservedMemoryPerPartition")
.categories("worker")
@@ -2531,6 +2534,16 @@ object CelebornConf extends Logging {
.checkValue(v => v < Int.MaxValue, "Reserved memory per partition must
be less than 2GB.")
.createWithDefaultString("1mb")
+ val WORKER_PARTITION_SORTER_RESERVED_MEMORY_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.worker.sortPartition.reservedMemory.enabled")
+ .categories("worker")
+ .doc(s"When true, partition sorter will reserve memory configured by
`${WORKER_PARTITION_SORTER_RESERVED_MEMORY_PER_PARTITION.key}` to allocate a
block of memory for warming up " +
+ "while sorting a shuffle file off-heap with page cache for non-hdfs
files." +
+ "Otherwise, partition sorter seeks to position of each block and does
not warm up for non-hdfs files.")
+ .version("0.5.0")
+ .booleanConf
+ .createWithDefault(true)
+
val WORKER_FLUSHER_BUFFER_SIZE: ConfigEntry[Long] =
buildConf("celeborn.worker.flusher.buffer.size")
.categories("worker")
@@ -2747,7 +2760,7 @@ object CelebornConf extends Logging {
.intConf
.createWithDefault(3)
- val PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD: ConfigEntry[Double] =
+ val WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD:
ConfigEntry[Double] =
buildConf("celeborn.worker.partitionSorter.directMemoryRatioThreshold")
.categories("worker")
.doc("Max ratio of partition sorter's memory for sorting, when reserved
memory is higher than max partition " +
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index 1f7f48a24..a6773fbfd 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -112,6 +112,7 @@ license: |
| celeborn.worker.shuffle.partitionSplit.min | 1m | Min size for a partition
to split | 0.3.0 | celeborn.shuffle.partitionSplit.min |
| celeborn.worker.sortPartition.indexCache.expire | 180s | PartitionSorter's
cache item expire time. | 0.4.0 | |
| celeborn.worker.sortPartition.indexCache.maxWeight | 100000 |
PartitionSorter's cache max weight for index buffer. | 0.4.0 | |
+| celeborn.worker.sortPartition.reservedMemory.enabled | true | When true,
partition sorter will reserve memory configured by
`celeborn.worker.sortPartition.reservedMemoryPerPartition` to allocate a block
of memory for warming up while sorting a shuffle file off-heap with page cache
for non-hdfs files.Otherwise, partition sorter seeks to position of each block
and does not warm up for non-hdfs files. | 0.5.0 | |
| celeborn.worker.sortPartition.reservedMemoryPerPartition | 1mb | Reserved
memory when sorting a shuffle file off-heap. | 0.3.0 |
celeborn.worker.partitionSorter.reservedMemoryPerPartition |
| celeborn.worker.sortPartition.threads | <undefined> |
PartitionSorter's thread counts. It's recommended to set at least `64` when
`HDFS` is enabled in `celeborn.storage.activeTypes`. | 0.3.0 |
celeborn.worker.partitionSorter.threads |
| celeborn.worker.sortPartition.timeout | 220s | Timeout for a shuffle file to
sort. | 0.3.0 | celeborn.worker.partitionSorter.sort.timeout |
diff --git
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java
index ec6a0d065..c58ccc55e 100644
---
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java
+++
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java
@@ -108,7 +108,7 @@ public class MemoryManager {
double pausePushDataRatio = conf.workerDirectMemoryRatioToPauseReceive();
double pauseReplicateRatio =
conf.workerDirectMemoryRatioToPauseReplicate();
double resumeRatio = conf.workerDirectMemoryRatioToResume();
- double maxSortMemRatio = conf.partitionSorterDirectMemoryRatioThreshold();
+ double maxSortMemRatio =
conf.workerPartitionSorterDirectMemoryRatioThreshold();
double readBufferRatio = conf.workerDirectMemoryRatioForReadBuffer();
double shuffleStorageRatio =
conf.workerDirectMemoryRatioForShuffleStorage();
long checkInterval = conf.workerDirectMemoryPressureCheckIntervalMs();
diff --git
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
index 5e4a9aee8..0c376aec5 100644
---
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
+++
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
@@ -86,6 +86,7 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
private final AtomicLong sortedFilesSize = new AtomicLong();
protected final long sortTimeout;
protected final long shuffleChunkSize;
+ protected final boolean reservedMemoryEnabled;
protected final long reservedMemoryPerPartition;
private final boolean gracefulShutdown;
private final long partitionSorterShutdownAwaitTime;
@@ -99,12 +100,13 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
public PartitionFilesSorter(
MemoryManager memoryManager, CelebornConf conf, AbstractSource source) {
- this.sortTimeout = conf.partitionSorterSortPartitionTimeout();
+ this.sortTimeout = conf.workerPartitionSorterSortPartitionTimeout();
this.shuffleChunkSize = conf.shuffleChunkSize();
- this.reservedMemoryPerPartition =
conf.partitionSorterReservedMemoryPerPartition();
+ this.reservedMemoryEnabled =
conf.workerPartitionSorterReservedMemoryEnabled();
+ this.reservedMemoryPerPartition =
conf.workerPartitionSorterReservedMemoryPerPartition();
this.partitionSorterShutdownAwaitTime =
conf.workerGracefulShutdownPartitionSorterCloseAwaitTimeMs();
- this.indexCacheMaxWeight = conf.partitionSorterIndexCacheMaxWeight();
+ this.indexCacheMaxWeight = conf.workerPartitionSorterIndexCacheMaxWeight();
this.source = source;
this.cleaner = new PartitionFilesCleaner(this);
this.gracefulShutdown = conf.workerGracefulShutdown();
@@ -129,12 +131,12 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
fileSorterExecutors =
ThreadUtils.newDaemonCachedThreadPool(
- "worker-file-sorter-executor", conf.partitionSorterThreads(), 120);
+ "worker-file-sorter-executor",
conf.workerPartitionSorterThreads(), 120);
indexCache =
CacheBuilder.newBuilder()
- .concurrencyLevel(conf.partitionSorterThreads())
- .expireAfterAccess(conf.partitionSorterIndexExpire(),
TimeUnit.MILLISECONDS)
+ .concurrencyLevel(conf.workerPartitionSorterThreads())
+ .expireAfterAccess(conf.workerPartitionSorterIndexExpire(),
TimeUnit.MILLISECONDS)
.maximumWeight(indexCacheMaxWeight)
.weigher(
(key, cache) ->
@@ -149,9 +151,11 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
try {
while (!shutdown) {
FileSorter task = shuffleSortTaskDeque.take();
- memoryManager.reserveSortMemory(reservedMemoryPerPartition);
- while (!memoryManager.sortMemoryReady()) {
- Thread.sleep(20);
+ if (task.isWarmUp) {
+ memoryManager.reserveSortMemory(reservedMemoryPerPartition);
+ while (!memoryManager.sortMemoryReady()) {
+ Thread.sleep(20);
+ }
}
fileSorterExecutors.submit(
() -> {
@@ -160,7 +164,9 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
} catch (InterruptedException e) {
logger.warn("File sorter thread was interrupted.");
} finally {
-
memoryManager.releaseSortMemory(reservedMemoryPerPartition);
+ if (task.isWarmUp) {
+
memoryManager.releaseSortMemory(reservedMemoryPerPartition);
+ }
}
});
}
@@ -545,6 +551,7 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
private final String fileId;
private final String shuffleKey;
private final boolean isHdfs;
+ private final boolean isWarmUp;
private final FileInfo originFileInfo;
private FSDataInputStream hdfsOriginInput = null;
@@ -557,6 +564,7 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
this.originFilePath = fileInfo.getFilePath();
this.sortedFilePath = Utils.getSortedFilePath(originFilePath);
this.isHdfs = fileInfo.isHdfs();
+ this.isWarmUp = !isHdfs && reservedMemoryEnabled;
this.originFileLen = fileInfo.getFileLength();
this.fileId = fileId;
this.shuffleKey = shuffleKey;
@@ -590,9 +598,9 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
Map<Integer, List<ShuffleBlockInfo>> sortedBlockInfoMap = new
HashMap<>();
int batchHeaderLen = 16;
- int reserveMemory = (int) reservedMemoryPerPartition;
ByteBuffer headerBuf = ByteBuffer.allocate(batchHeaderLen);
- ByteBuffer paddingBuf = ByteBuffer.allocateDirect(reserveMemory);
+ ByteBuffer paddingBuf =
+ isWarmUp ? ByteBuffer.allocateDirect((int)
reservedMemoryPerPartition) : null;
long index = 0;
while (index != originFileLen) {
@@ -612,7 +620,6 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
singleMapIdShuffleBlockList.add(blockInfo);
index += batchHeaderLen + compressedSize;
- paddingBuf.clear();
readBufferBySize(paddingBuf, compressedSize);
}
@@ -736,10 +743,13 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
private void readBufferBySize(ByteBuffer buffer, int toRead) throws
IOException {
if (isHdfs) {
- // HDFS don't need warmup
+ // HDFS does not need to warm up.
hdfsOriginInput.seek(toRead + hdfsOriginInput.getPos());
- } else {
+ } else if (reservedMemoryEnabled) {
+ buffer.clear();
readChannelBySize(originFileChannel, buffer, originFilePath, toRead);
+ } else {
+ originFileChannel.position(toRead + originFileChannel.position());
}
}
}
diff --git
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManagerSuiteJ.java
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManagerSuiteJ.java
index ad9b1e311..4050691b1 100644
---
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManagerSuiteJ.java
+++
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManagerSuiteJ.java
@@ -53,7 +53,7 @@ public class CreditStreamManagerSuiteJ {
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_RECEIVE().key(),
"0.8");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_REPLICATE().key(),
"0.9");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_RESUME().key(), "0.5");
-
conf.set(CelebornConf.PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
+
conf.set(CelebornConf.WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_READ_BUFFER().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_SHUFFLE_STORAGE().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_CHECK_INTERVAL().key(), "10");
diff --git
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataWriterSuiteJ.java
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataWriterSuiteJ.java
index a201018e1..bc7772001 100644
---
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataWriterSuiteJ.java
+++
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataWriterSuiteJ.java
@@ -98,7 +98,7 @@ public class MapPartitionDataWriterSuiteJ {
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_RECEIVE().key(),
"0.8");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_REPLICATE().key(),
"0.9");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_RESUME().key(), "0.5");
-
conf.set(CelebornConf.PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
+
conf.set(CelebornConf.WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_READ_BUFFER().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_SHUFFLE_STORAGE().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_CHECK_INTERVAL().key(), "10");
diff --git
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorterSuiteJ.java
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorterSuiteJ.java
index 3e3862468..cce7e1f29 100644
---
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorterSuiteJ.java
+++
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorterSuiteJ.java
@@ -111,7 +111,7 @@ public class PartitionFilesSorterSuiteJ {
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_RECEIVE().key(),
"0.8");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_REPLICATE().key(),
"0.9");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_RESUME().key(), "0.5");
-
conf.set(CelebornConf.PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
+
conf.set(CelebornConf.WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_READ_BUFFER().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_SHUFFLE_STORAGE().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_CHECK_INTERVAL().key(), "10");
diff --git
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/ReducePartitionDataWriterSuiteJ.java
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/ReducePartitionDataWriterSuiteJ.java
index b9955cc5f..e6b688b43 100644
---
a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/ReducePartitionDataWriterSuiteJ.java
+++
b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/ReducePartitionDataWriterSuiteJ.java
@@ -129,7 +129,7 @@ public class ReducePartitionDataWriterSuiteJ {
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_RECEIVE().key(),
"0.8");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_PAUSE_REPLICATE().key(),
"0.9");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_RESUME().key(), "0.5");
-
conf.set(CelebornConf.PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
+
conf.set(CelebornConf.WORKER_PARTITION_SORTER_DIRECT_MEMORY_RATIO_THRESHOLD().key(),
"0.6");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_READ_BUFFER().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_RATIO_FOR_SHUFFLE_STORAGE().key(),
"0.1");
conf.set(CelebornConf.WORKER_DIRECT_MEMORY_CHECK_INTERVAL().key(), "10");
