This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 3da61597 [CELEBORN-326)] [Flink] lifecycleManager supports
flink-yarn-session mode to handle multiple Flink jobs. (#1260)
3da61597 is described below
commit 3da615972e970128119f3986da68e984ddbdbd01
Author: Shuang <[email protected]>
AuthorDate: Wed Feb 22 15:37:24 2023 +0800
[CELEBORN-326)] [Flink] lifecycleManager supports flink-yarn-session mode
to handle multiple Flink jobs. (#1260)
---
client-flink/flink-1.14/pom.xml | 11 +++
.../plugin/flink/FlinkResultPartitionInfo.java | 17 +++-
.../plugin/flink/RemoteShuffleDescriptor.java | 31 +++++--
.../plugin/flink/RemoteShuffleInputGate.java | 2 +-
.../celeborn/plugin/flink/RemoteShuffleMaster.java | 98 ++++++++++++----------
.../plugin/flink/RemoteShuffleOutputGate.java | 2 +-
.../plugin/flink/RemoteShuffleResource.java | 10 +++
.../celeborn/plugin/flink/utils/FlinkUtils.java | 4 +-
.../plugin/flink/RemoteShuffleMasterTest.java | 43 +++++++++-
.../flink-1.14/src/test/resources/log4j2-test.xml | 41 +++++++++
.../apache/celeborn/client/LifecycleManager.scala | 2 +-
11 files changed, 200 insertions(+), 61 deletions(-)
diff --git a/client-flink/flink-1.14/pom.xml b/client-flink/flink-1.14/pom.xml
index eb9c90be..ffa10ba4 100644
--- a/client-flink/flink-1.14/pom.xml
+++ b/client-flink/flink-1.14/pom.xml
@@ -49,6 +49,7 @@
<artifactId>flink-runtime</artifactId>
<scope>provided</scope>
</dependency>
+ <!-- Test dependencies -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
@@ -59,5 +60,15 @@
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
index 588d0c7d..1d8d8185 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
@@ -17,6 +17,7 @@
package org.apache.celeborn.plugin.flink;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.shuffle.PartitionDescriptor;
import org.apache.flink.runtime.shuffle.ProducerDescriptor;
@@ -24,11 +25,13 @@ import org.apache.flink.runtime.shuffle.ProducerDescriptor;
import org.apache.celeborn.plugin.flink.utils.FlinkUtils;
public class FlinkResultPartitionInfo {
+ private final JobID jobID;
private final PartitionDescriptor partitionDescriptor;
private final ProducerDescriptor producerDescriptor;
public FlinkResultPartitionInfo(
- PartitionDescriptor partitionDescriptor, ProducerDescriptor
producerDescriptor) {
+ JobID jobId, PartitionDescriptor partitionDescriptor, ProducerDescriptor
producerDescriptor) {
+ this.jobID = jobId;
this.partitionDescriptor = partitionDescriptor;
this.producerDescriptor = producerDescriptor;
}
@@ -39,7 +42,7 @@ public class FlinkResultPartitionInfo {
}
public String getShuffleId() {
- return FlinkUtils.toShuffleId(partitionDescriptor.getResultId());
+ return FlinkUtils.toShuffleId(jobID, partitionDescriptor.getResultId());
}
public int getTaskId() {
@@ -49,4 +52,14 @@ public class FlinkResultPartitionInfo {
public String getAttemptId() {
return FlinkUtils.toAttemptId(producerDescriptor.getProducerExecutionId());
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("FlinkResultPartitionInfo{");
+ sb.append("jobID=").append(jobID);
+ sb.append(",
partitionDescriptor=").append(partitionDescriptor.getPartitionId());
+ sb.append(",
producerDescriptor=").append(producerDescriptor.getProducerExecutionId());
+ sb.append('}');
+ return sb.toString();
+ }
}
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleDescriptor.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleDescriptor.java
index 30e050f2..0678d4d2 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleDescriptor.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleDescriptor.java
@@ -19,22 +19,24 @@ package org.apache.celeborn.plugin.flink;
import java.util.Optional;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
public class RemoteShuffleDescriptor implements ShuffleDescriptor {
-
+ private final String celebornAppId;
+ // jobId-datasetId
+ private final String shuffleId;
private final ResultPartitionID resultPartitionID;
-
- private final JobID jobID;
-
private final RemoteShuffleResource shuffleResource;
public RemoteShuffleDescriptor(
- JobID jobID, ResultPartitionID resultPartitionID, RemoteShuffleResource
shuffleResource) {
- this.jobID = jobID;
+ String celebornAppId,
+ String shuffleId,
+ ResultPartitionID resultPartitionID,
+ RemoteShuffleResource shuffleResource) {
+ this.celebornAppId = celebornAppId;
+ this.shuffleId = shuffleId;
this.resultPartitionID = resultPartitionID;
this.shuffleResource = shuffleResource;
}
@@ -44,8 +46,8 @@ public class RemoteShuffleDescriptor implements
ShuffleDescriptor {
return resultPartitionID;
}
- public JobID getJobID() {
- return jobID;
+ public String getCelebornAppId() {
+ return celebornAppId;
}
public RemoteShuffleResource getShuffleResource() {
@@ -56,4 +58,15 @@ public class RemoteShuffleDescriptor implements
ShuffleDescriptor {
public Optional<ResourceID> storesLocalResourcesOn() {
return Optional.empty();
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("RemoteShuffleDescriptor{");
+ sb.append("celebornAppId='").append(celebornAppId).append('\'');
+ sb.append(", shuffleId='").append(shuffleId).append('\'');
+ sb.append(", resultPartitionID=").append(resultPartitionID);
+ sb.append(", shuffleResource=").append(shuffleResource);
+ sb.append('}');
+ return sb.toString();
+ }
}
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
index 07472f40..1192125c 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleInputGate.java
@@ -185,7 +185,7 @@ public class RemoteShuffleInputGate extends
IndexedInputGate {
remoteDescriptor.getShuffleResource().getMapPartitionShuffleDescriptor();
LOG.debug("create shuffle reader for descriptor {}", shuffleDescriptor);
- String applicationId = remoteDescriptor.getJobID().toString();
+ String applicationId = remoteDescriptor.getCelebornAppId();
RemoteBufferStreamReader reader =
new RemoteBufferStreamReader(
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
index 222c0516..1a8d3d2b 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java
@@ -17,11 +17,11 @@
package org.apache.celeborn.plugin.flink;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import org.apache.flink.api.common.JobID;
@@ -41,10 +41,12 @@ import org.apache.celeborn.plugin.flink.utils.ThreadUtils;
public class RemoteShuffleMaster implements
ShuffleMaster<RemoteShuffleDescriptor> {
private static final Logger LOG =
LoggerFactory.getLogger(RemoteShuffleMaster.class);
-
private final ShuffleMasterContext shuffleMasterContext;
- // Flink JobId -> Celeborn LifecycleManager
- private Map<JobID, LifecycleManager> lifecycleManagers = new
ConcurrentHashMap<>();
+ // Flink JobId -> Celeborn register shuffleIds
+ private Map<JobID, Set<Integer>> jobShuffleIds = new ConcurrentHashMap<>();
+ private String celebornAppId;
+ private volatile LifecycleManager lifecycleManager;
+
private final ScheduledThreadPoolExecutor executor =
new ScheduledThreadPoolExecutor(
1,
@@ -57,46 +59,43 @@ public class RemoteShuffleMaster implements
ShuffleMaster<RemoteShuffleDescripto
@Override
public void registerJob(JobShuffleContext context) {
- JobID jobId = context.getJobId();
- Future<?> submit =
- executor.submit(
- () -> {
- if (lifecycleManagers.containsKey(jobId)) {
- throw new RuntimeException("Duplicated registration job: " +
jobId);
- } else {
- CelebornConf celebornConf =
-
FlinkUtils.toCelebornConf(shuffleMasterContext.getConfiguration());
- LifecycleManager lifecycleManager =
- new LifecycleManager(FlinkUtils.toCelebornAppId(jobId),
celebornConf);
- lifecycleManagers.put(jobId, lifecycleManager);
- }
- });
+ JobID jobID = context.getJobId();
+ if (lifecycleManager == null) {
+ synchronized (RemoteShuffleMaster.class) {
+ if (lifecycleManager == null) {
+ // use first jobID as celeborn shared appId for all other flink jobs
+ celebornAppId = FlinkUtils.toCelebornAppId(jobID);
+ CelebornConf celebornConf =
+
FlinkUtils.toCelebornConf(shuffleMasterContext.getConfiguration());
+ lifecycleManager = new LifecycleManager(celebornAppId, celebornConf);
+ }
+ }
+ }
- try {
- submit.get();
- } catch (InterruptedException e) {
- LOG.error("Encounter interruptedException when registration job: {}.",
jobId, e);
- throw new RuntimeException(e);
- } catch (ExecutionException e) {
- LOG.error("Encounter an error when registration job: {}.", jobId, e);
- throw new RuntimeException(e.getCause());
+ Set<Integer> previousShuffleIds = jobShuffleIds.putIfAbsent(jobID, new
HashSet<>());
+ if (previousShuffleIds != null) {
+ throw new RuntimeException("Duplicated registration job: " + jobID);
}
}
@Override
public void unregisterJob(JobID jobID) {
- executor.execute(
- () -> {
- try {
- LOG.info("Unregister job: {}.", jobID);
- LifecycleManager lifecycleManager =
lifecycleManagers.remove(jobID);
- if (lifecycleManager != null) {
- lifecycleManager.stop();
+ LOG.info("Unregister job: {}.", jobID);
+ Set<Integer> shuffleIds = jobShuffleIds.remove(jobID);
+ if (shuffleIds != null) {
+ executor.execute(
+ () -> {
+ try {
+ synchronized (shuffleIds) {
+ for (Integer shuffleId : shuffleIds) {
+ lifecycleManager.handleUnregisterShuffle(celebornAppId,
shuffleId);
+ }
+ }
+ } catch (Throwable throwable) {
+ LOG.error("Encounter an error when unregistering job: {}.",
jobID, throwable);
}
- } catch (Throwable throwable) {
- LOG.error("Encounter an error when unregistering job: {}.", jobID,
throwable);
- }
- });
+ });
+ }
}
@Override
@@ -105,14 +104,23 @@ public class RemoteShuffleMaster implements
ShuffleMaster<RemoteShuffleDescripto
CompletableFuture<RemoteShuffleDescriptor> completableFuture =
CompletableFuture.supplyAsync(
() -> {
- LifecycleManager lifecycleManager = lifecycleManagers.get(jobID);
+ Set<Integer> shuffleIds = jobShuffleIds.get(jobID);
+ if (shuffleIds == null) {
+ throw new RuntimeException("Can not find job in
lifecycleManager, job: " + jobID);
+ }
+
FlinkResultPartitionInfo resultPartitionInfo =
- new FlinkResultPartitionInfo(partitionDescriptor,
producerDescriptor);
+ new FlinkResultPartitionInfo(jobID, partitionDescriptor,
producerDescriptor);
LifecycleManager.ShuffleTask shuffleTask =
lifecycleManager.encodeExternalShuffleTask(
resultPartitionInfo.getShuffleId(),
resultPartitionInfo.getTaskId(),
resultPartitionInfo.getAttemptId());
+
+ synchronized (shuffleIds) {
+ shuffleIds.add(shuffleTask.shuffleId());
+ }
+
ShuffleResourceDescriptor shuffleResourceDescriptor =
new ShuffleResourceDescriptor(shuffleTask);
RemoteShuffleResource remoteShuffleResource =
@@ -121,7 +129,10 @@ public class RemoteShuffleMaster implements
ShuffleMaster<RemoteShuffleDescripto
lifecycleManager.getRssMetaServicePort(),
shuffleResourceDescriptor);
return new RemoteShuffleDescriptor(
- jobID, resultPartitionInfo.getResultPartitionId(),
remoteShuffleResource);
+ celebornAppId,
+ resultPartitionInfo.getShuffleId(),
+ resultPartitionInfo.getResultPartitionId(),
+ remoteShuffleResource);
},
executor);
@@ -136,9 +147,8 @@ public class RemoteShuffleMaster implements
ShuffleMaster<RemoteShuffleDescripto
@Override
public void close() throws Exception {
try {
- for (LifecycleManager lifecycleManager : lifecycleManagers.values()) {
- lifecycleManager.stop();
- }
+ jobShuffleIds.clear();
+ lifecycleManager.stop();
} catch (Exception e) {
LOG.warn("Encounter exception when shutdown: " + e.getMessage(), e);
}
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
index 52a9a249..247b22d9 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleOutputGate.java
@@ -98,7 +98,7 @@ public class RemoteShuffleOutputGate {
this.celebornConf = celebornConf;
this.numMappers = numMappers;
this.bufferSize = bufferSize;
- this.applicationId = shuffleDesc.getJobID().toString();
+ this.applicationId = shuffleDesc.getCelebornAppId();
this.shuffleId =
shuffleDesc.getShuffleResource().getMapPartitionShuffleDescriptor().getShuffleId();
this.mapId =
shuffleDesc.getShuffleResource().getMapPartitionShuffleDescriptor().getMapId();
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResource.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResource.java
index ad8eef06..e57ac249 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResource.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResource.java
@@ -47,4 +47,14 @@ public class RemoteShuffleResource implements
ShuffleResource {
public int getRssMetaServicePort() {
return rssMetaServicePort;
}
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("RemoteShuffleResource{");
+ sb.append("rssMetaServiceHost='").append(rssMetaServiceHost).append('\'');
+ sb.append(", rssMetaServicePort=").append(rssMetaServicePort);
+ sb.append(",
shuffleResourceDescriptor=").append(shuffleResourceDescriptor);
+ sb.append('}');
+ return sb.toString();
+ }
}
diff --git
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/utils/FlinkUtils.java
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/utils/FlinkUtils.java
index 1182c726..06cc3856 100644
---
a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/utils/FlinkUtils.java
+++
b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/utils/FlinkUtils.java
@@ -45,8 +45,8 @@ public class FlinkUtils {
return jobID.toString();
}
- public static String toShuffleId(IntermediateDataSetID dataSetID) {
- return dataSetID.toString();
+ public static String toShuffleId(JobID jobID, IntermediateDataSetID
dataSetID) {
+ return jobID.toString() + "-" + dataSetID.toString();
}
public static String toAttemptId(ExecutionAttemptID attemptID) {
diff --git
a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
index 20410052..873693d1 100644
---
a/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
+++
b/client-flink/flink-1.14/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterTest.java
@@ -69,6 +69,7 @@ public class RemoteShuffleMasterTest {
// unRegister job
remoteShuffleMaster.unregisterJob(jobShuffleContext.getJobId());
+ remoteShuffleMaster.registerJob(jobShuffleContext);
}
@Test
@@ -88,7 +89,8 @@ public class RemoteShuffleMasterTest {
ShuffleResource shuffleResource =
remoteShuffleDescriptor.getShuffleResource();
ShuffleResourceDescriptor mapPartitionShuffleDescriptor =
shuffleResource.getMapPartitionShuffleDescriptor();
- System.out.println(mapPartitionShuffleDescriptor.toString());
+
+ LOG.info("remoteShuffleDescriptor:", remoteShuffleDescriptor);
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getShuffleId());
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getPartitionId());
Assert.assertEquals(0, mapPartitionShuffleDescriptor.getAttemptId());
@@ -120,6 +122,45 @@ public class RemoteShuffleMasterTest {
Assert.assertEquals(1, mapPartitionShuffleDescriptor.getMapId());
}
+ @Test
+ public void testRegisterMultipleJobs()
+ throws UnknownHostException, ExecutionException, InterruptedException {
+ JobID jobID1 = JobID.generate();
+ JobShuffleContext jobShuffleContext1 = createJobShuffleContext(jobID1);
+ remoteShuffleMaster.registerJob(jobShuffleContext1);
+
+ JobID jobID2 = JobID.generate();
+ JobShuffleContext jobShuffleContext2 = createJobShuffleContext(jobID2);
+ remoteShuffleMaster.registerJob(jobShuffleContext2);
+
+ IntermediateDataSetID intermediateDataSetID = new IntermediateDataSetID();
+ PartitionDescriptor partitionDescriptor =
createPartitionDescriptor(intermediateDataSetID, 0);
+ ProducerDescriptor producerDescriptor = createProducerDescriptor();
+ RemoteShuffleDescriptor remoteShuffleDescriptor1 =
+ remoteShuffleMaster
+ .registerPartitionWithProducer(jobID1, partitionDescriptor,
producerDescriptor)
+ .get();
+
+ // use same datasetId but different jobId
+ RemoteShuffleDescriptor remoteShuffleDescriptor2 =
+ remoteShuffleMaster
+ .registerPartitionWithProducer(jobID2, partitionDescriptor,
producerDescriptor)
+ .get();
+
+ Assert.assertEquals(
+ remoteShuffleDescriptor1
+ .getShuffleResource()
+ .getMapPartitionShuffleDescriptor()
+ .getShuffleId(),
+ 0);
+ Assert.assertEquals(
+ remoteShuffleDescriptor2
+ .getShuffleResource()
+ .getMapPartitionShuffleDescriptor()
+ .getShuffleId(),
+ 1);
+ }
+
@After
public void tearDown() {
if (remoteShuffleMaster != null) {
diff --git a/client-flink/flink-1.14/src/test/resources/log4j2-test.xml
b/client-flink/flink-1.14/src/test/resources/log4j2-test.xml
new file mode 100644
index 00000000..e91e2f31
--- /dev/null
+++ b/client-flink/flink-1.14/src/test/resources/log4j2-test.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<Configuration status="INFO">
+ <Appenders>
+ <Console name="stdout" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{yy/MM/dd HH:mm:ss,SSS} %p [%t] %c{1}:
%m%n%ex"/>
+ <Filters>
+ <ThresholdFilter level="FATAL"/>
+ </Filters>
+ </Console>
+ <File name="file" fileName="target/unit-tests.log">
+ <PatternLayout pattern="%d{yy/MM/dd HH:mm:ss,SSS} %p [%t] %c{1}:
%m%n%ex"/>
+ </File>
+ </Appenders>
+ <Loggers>
+ <Root level="INFO">
+ <AppenderRef ref="stdout"/>
+ <AppenderRef ref="file"/>
+ </Root>
+ <Logger name="org.sparkproject.jetty" level="WARN" additivity="false">
+ <AppenderRef ref="stdout"/>
+ <AppenderRef ref="file"/>
+ </Logger>
+ </Loggers>
+</Configuration>
diff --git
a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
index 22ae9dc1..21b9ac5c 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -682,7 +682,7 @@ class LifecycleManager(appId: String, val conf:
CelebornConf) extends RpcEndpoin
reply(mapperAttemptFinishedSuccess)
}
- private def handleUnregisterShuffle(
+ def handleUnregisterShuffle(
appId: String,
shuffleId: Int): Unit = {
if (getPartitionType(shuffleId) == PartitionType.REDUCE) {
