This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch branch-1.8
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/branch-1.8 by this push:
new e55a1fcaf3 [GLUTEN-13276][VL] Fix CelebornShuffleManager init fails
with NoSuchMethodException: FailedShuffleCleaner.<init>() (#13283)
e55a1fcaf3 is described below
commit e55a1fcaf35937da0b2e3bdb30d0ccb40ca03cdd
Author: maoyuwei1 <[email protected]>
AuthorDate: Tue Oct 27 17:13:26 2026 +0800
[GLUTEN-13276][VL] Fix CelebornShuffleManager init fails with
NoSuchMethodException: FailedShuffleCleaner.<init>() (#13283)
---
gluten-celeborn/pom.xml | 17 ++
.../gluten/celeborn/CelebornShuffleManager.java | 4 +-
.../shuffle/gluten/celeborn/CelebornUtils.java | 52 ++++--
.../celeborn/CelebornShuffleManagerTest.java | 191 ++++++++++++++++++++
.../shuffle/gluten/celeborn/CelebornUtilsTest.java | 200 +++++++++++++++++++++
.../celeborn/FailedShuffleCleanerTestSupport.java | 57 ++++++
...TestCelebornColumnarBatchSerializerFactory.java | 32 ++++
.../celeborn/TestCelebornShuffleWriterFactory.java | 43 +++++
...celeborn.CelebornColumnarBatchSerializerFactory | 1 +
...le.gluten.celeborn.CelebornShuffleWriterFactory | 1 +
10 files changed, 586 insertions(+), 12 deletions(-)
diff --git a/gluten-celeborn/pom.xml b/gluten-celeborn/pom.xml
index 997de509bf..4cd255b69b 100755
--- a/gluten-celeborn/pom.xml
+++ b/gluten-celeborn/pom.xml
@@ -67,6 +67,23 @@
<version>${guava.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-client</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-inline</artifactId>
+ <version>4.11.0</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManager.java
b/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManager.java
index 0a82e32726..a3ba5d6dd0 100644
---
a/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManager.java
+++
b/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManager.java
@@ -179,8 +179,8 @@ public class CelebornShuffleManager
// for Celeborn 0.6.0
CelebornUtils.incrementApplicationCount(lifecycleManager);
CelebornUtils.registerCancelShuffleCallback(lifecycleManager);
- CelebornUtils.stageRerun(
- stageRerunEnabled, lifecycleManager, celebornConf,
failedShuffleCleaner);
+ failedShuffleCleaner =
+ CelebornUtils.stageRerun(stageRerunEnabled, lifecycleManager,
celebornConf);
shuffleClient =
CelebornUtils.getShuffleClient(
diff --git
a/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtils.java
b/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtils.java
index 292ed9cbcd..98a1701d08 100644
---
a/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtils.java
+++
b/gluten-celeborn/src/main/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtils.java
@@ -200,9 +200,20 @@ public class CelebornUtils {
try {
Class<?> clazz = Class.forName(className);
- Constructor<?> constructor = clazz.getConstructor();
+ Constructor<?> matched = null;
+ for (Constructor<?> constructor : clazz.getConstructors()) {
+ if (isConstructorMatched(constructor, args)) {
+ matched = constructor;
+ break;
+ }
+ }
+ if (matched == null) {
+ throw new NoSuchMethodException(
+ className + ".<init> with " + args.length + " matched
argument(s)");
+ }
+ matched.setAccessible(true);
- return constructor.newInstance(args);
+ return matched.newInstance(args);
} catch (ClassNotFoundException e) {
return null;
@@ -212,6 +223,23 @@ public class CelebornUtils {
}
}
+ private static boolean isConstructorMatched(Constructor<?> constructor,
Object... args) {
+ Class<?>[] parameterTypes = constructor.getParameterTypes();
+ if (parameterTypes.length != args.length) {
+ return false;
+ }
+ for (int i = 0; i < parameterTypes.length; i++) {
+ if (args[i] == null) {
+ if (parameterTypes[i].isPrimitive()) {
+ return false;
+ }
+ } else if (!parameterTypes[i].isAssignableFrom(args[i].getClass())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
public static CelebornShuffleHandle getCelebornShuffleHandle(
String appUniqueId,
String lifecycleManagerHost,
@@ -359,11 +387,9 @@ public class CelebornUtils {
}
}
- public static void stageRerun(
- boolean stageRerunEnabled,
- LifecycleManager lifecycleManager,
- CelebornConf celebornConf,
- Object failedShuffleCleaner) {
+ public static Object stageRerun(
+ boolean stageRerunEnabled, LifecycleManager lifecycleManager,
CelebornConf celebornConf) {
+ Object failedShuffleCleaner = null;
if (stageRerunEnabled) {
MapOutputTrackerMaster mapOutputTracker =
(MapOutputTrackerMaster) SparkEnv.get().mapOutputTracker();
@@ -375,9 +401,10 @@ public class CelebornUtils {
// for Celeborn 0.6.0
registerCelebornSkewShuffleCheckCallback(lifecycleManager, celebornConf);
- fetchCleanFailedShuffle(lifecycleManager, celebornConf,
failedShuffleCleaner);
+ failedShuffleCleaner = fetchCleanFailedShuffle(lifecycleManager,
celebornConf);
reducerFileGroupBroadcast(lifecycleManager, celebornConf);
}
+ return failedShuffleCleaner;
}
public static void registerShuffleTrackerCallback(
@@ -538,14 +565,18 @@ public class CelebornUtils {
}
}
- public static void fetchCleanFailedShuffle(
- LifecycleManager lifecycleManager, CelebornConf celebornConf, Object
failedShuffleCleaner) {
+ public static Object fetchCleanFailedShuffle(
+ LifecycleManager lifecycleManager, CelebornConf celebornConf) {
+ Object failedShuffleCleaner = null;
try {
// for Celeborn 0.6.0
Method clientFetchCleanFailedShuffleMethod =
celebornConf.getClass().getDeclaredMethod("clientFetchCleanFailedShuffle");
if ((Boolean) clientFetchCleanFailedShuffleMethod.invoke(celebornConf)) {
failedShuffleCleaner = createInstance(FAILED_SHUFFLE_CLEANER_NAME,
lifecycleManager);
+ if (failedShuffleCleaner == null) {
+ return null;
+ }
Method registerValidateCelebornShuffleIdForCleanCallbackMethod =
lifecycleManager
.getClass()
@@ -589,6 +620,7 @@ public class CelebornUtils {
} catch (Exception e) {
throw new RuntimeException(e);
}
+ return failedShuffleCleaner;
}
public static void reducerFileGroupBroadcast(
diff --git
a/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManagerTest.java
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManagerTest.java
new file mode 100644
index 0000000000..007d761363
--- /dev/null
+++
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornShuffleManagerTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.spark.shuffle.gluten.celeborn;
+
+import org.apache.celeborn.client.LifecycleManager;
+import org.apache.celeborn.client.ShuffleClient;
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.shuffle.celeborn.SparkUtils;
+import org.junit.Test;
+import org.mockito.Answers;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.stubbing.Answer;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+/**
+ * Tests for the {@code FailedShuffleCleaner} ownership of {@link
CelebornShuffleManager}.
+ *
+ * <p>The test is hermetic: {@link LifecycleManager} construction and the
{@link ShuffleClient}
+ * factory are mocked, so no {@code RpcEnv}, no {@code ApplicationHeartbeater}
and no RPC against a
+ * Celeborn master is involved. Only the cleaner itself is a real Celeborn
object, because its
+ * executor is the resource whose lifecycle is under test.
+ */
+public class CelebornShuffleManagerTest {
+
+ private static final String STAGE_RERUN_KEY =
"spark.celeborn.client.spark.stageRerun.enabled";
+ private static final String CLEAN_FAILED_SHUFFLE_KEY =
+ "spark.celeborn.client.spark.fetch.cleanFailedShuffle";
+ private static final String CLEAN_FAILED_SHUFFLE_INTERVAL_KEY =
+ "spark.celeborn.client.spark.fetch.cleanFailedShuffleInterval";
+
+ @Test
+ public void
testManagerRetainsFailedShuffleCleanerAndStopShutsDownItsExecutor() throws
Exception {
+ FailedShuffleCleanerTestSupport.assumeSupported();
+
+ // SparkConf(false) ignores ambient spark.* system properties, so the
manager only sees the
+ // settings below.
+ SparkConf sparkConf =
+ new SparkConf(false)
+ .set(STAGE_RERUN_KEY, "true")
+ .set(CLEAN_FAILED_SHUFFLE_KEY, "true")
+ // Long enough that the periodic cleanup never fires while the
test runs.
+ .set(CLEAN_FAILED_SHUFFLE_INTERVAL_KEY, "1h")
+ .set("spark.sql.adaptive.localShuffleReader.enabled", "false");
+
+ SparkEnv previousEnv = SparkEnv.get();
+ SparkEnv env = mock(SparkEnv.class);
+ when(env.executorId()).thenReturn("driver");
+
when(env.mapOutputTracker()).thenReturn(mock(MapOutputTrackerMaster.class));
+ SparkEnv.set(env);
+ try (MockedStatic<ShuffleClient> shuffleClients =
mockStatic(ShuffleClient.class);
+ MockedStatic<SparkUtils> sparkUtils = mockStatic(SparkUtils.class,
sparkUtilsAnswer());
+ MockedConstruction<LifecycleManager> lifecycleManagers =
+ mockConstruction(
+ LifecycleManager.class,
+ context ->
withSettings().defaultAnswer(lifecycleManagerAnswer(context)),
+ (lifecycleManager, context) -> {})) {
+ CelebornShuffleManager manager = new CelebornShuffleManager(sparkConf);
+ ExecutorService cleanerExecutor;
+ try {
+ initializeLifecycleManager(manager, "gluten-celeborn-test-app");
+ assertEquals(
+ "The driver side manager should create exactly one
LifecycleManager",
+ 1,
+ lifecycleManagers.constructed().size());
+
+ Object cleaner = failedShuffleCleanerOf(manager);
+ assertNotNull(
+ "CelebornShuffleManager should retain the FailedShuffleCleaner
created during"
+ + " LifecycleManager initialization",
+ cleaner);
+ assertEquals(CelebornUtils.FAILED_SHUFFLE_CLEANER_NAME,
cleaner.getClass().getName());
+
+ cleanerExecutor = FailedShuffleCleanerTestSupport.executorOf(cleaner);
+ assertNotNull("FailedShuffleCleaner should own a scheduled executor",
cleanerExecutor);
+ assertFalse(
+ "FailedShuffleCleaner executor should be running before the
manager is stopped",
+ cleanerExecutor.isShutdown());
+ } finally {
+ manager.stop();
+ }
+
+ assertTrue(
+ "CelebornShuffleManager.stop() should shut down the cleaner
executor",
+ cleanerExecutor.isShutdown());
+ assertTrue(
+ "cleaner executor should terminate once the manager is stopped",
+ cleanerExecutor.awaitTermination(10, TimeUnit.SECONDS));
+ assertNull(
+ "CelebornShuffleManager.stop() should release the cleaner reference",
+ failedShuffleCleanerOf(manager));
+ } finally {
+ SparkEnv.set(previousEnv);
+ }
+ }
+
+ /**
+ * Keeps the real Celeborn {@code SparkUtils} behavior, except for {@code
addSparkListener}, which
+ * dereferences {@code SparkContext.getActive()} without a null check and
would therefore require
+ * a live SparkContext. Registering the listener is unrelated to the cleaner
ownership under test.
+ */
+ private static Answer<Object> sparkUtilsAnswer() {
+ return invocation -> {
+ if ("addSparkListener".equals(invocation.getMethod().getName())) {
+ return null;
+ }
+ return invocation.callRealMethod();
+ };
+ }
+
+ /**
+ * Drives the production initialization path, which is where the cleaner has
to be retained. It is
+ * reached from {@code registerShuffle} in production, which additionally
needs a live
+ * SparkContext, so the private method is invoked directly instead.
+ */
+ private static void initializeLifecycleManager(CelebornShuffleManager
manager, String appId)
+ throws ReflectiveOperationException {
+ Method method =
+
CelebornShuffleManager.class.getDeclaredMethod("initializeLifecycleManager",
String.class);
+ method.setAccessible(true);
+ try {
+ method.invoke(manager, appId);
+ } catch (InvocationTargetException e) {
+ if (e.getCause() instanceof RuntimeException) {
+ throw (RuntimeException) e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ private static Object failedShuffleCleanerOf(CelebornShuffleManager manager)
+ throws ReflectiveOperationException {
+ Field field =
CelebornShuffleManager.class.getDeclaredField("failedShuffleCleaner");
+ field.setAccessible(true);
+ return field.get(manager);
+ }
+
+ /**
+ * Answers only what the production initialization path reads from the
LifecycleManager: the conf
+ * the manager passed to the constructor, and the counters it increments.
Everything else keeps
+ * the Mockito defaults, so no real LifecycleManager behavior is involved.
Return types are
+ * matched instead of method names to stay tolerant to Celeborn version
differences.
+ */
+ private static Answer<Object>
lifecycleManagerAnswer(MockedConstruction.Context context) {
+ CelebornConf celebornConf = (CelebornConf) context.arguments().get(1);
+ return invocation -> {
+ Class<?> returnType = invocation.getMethod().getReturnType();
+ if (CelebornConf.class.equals(returnType)) {
+ return celebornConf;
+ }
+ if (LongAdder.class.equals(returnType)) {
+ return new LongAdder();
+ }
+ return Answers.RETURNS_DEFAULTS.answer(invocation);
+ };
+ }
+}
diff --git
a/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtilsTest.java
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtilsTest.java
new file mode 100644
index 0000000000..915ba979da
--- /dev/null
+++
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/CelebornUtilsTest.java
@@ -0,0 +1,200 @@
+/*
+ * 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.spark.shuffle.gluten.celeborn;
+
+import org.apache.celeborn.client.LifecycleManager;
+import org.apache.celeborn.common.CelebornConf;
+import org.junit.Test;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the Celeborn reflection compatibility layer in {@link
CelebornUtils}. The manager side
+ * ownership of the cleaner is covered by {@link CelebornShuffleManagerTest}.
+ */
+public class CelebornUtilsTest {
+
+ /** Mirrors classes like {@code FailedShuffleCleaner} that only expose a
parameterized ctor. */
+ public static class ParameterizedOnly {
+ public final String value;
+
+ public ParameterizedOnly(String value) {
+ this.value = value;
+ }
+ }
+
+ /** Mirrors classes usable through both the no-arg and the parameterized
ctor. */
+ public static class NoArgAndParameterized {
+ public String value;
+
+ public NoArgAndParameterized() {
+ this.value = "no-arg";
+ }
+
+ public NoArgAndParameterized(String value) {
+ this.value = value;
+ }
+ }
+
+ /** Only accepts a primitive parameter, which can never be fed with a {@code
null} arg. */
+ public static class PrimitiveParameter {
+ public PrimitiveParameter(int value) {}
+ }
+
+ /** Accepts any reference type, so a {@code String} arg must be assignable.
*/
+ public static class SupertypeParameter {
+ public final Object value;
+
+ public SupertypeParameter(Object value) {
+ this.value = value;
+ }
+ }
+
+ @Test
+ public void testCreateInstanceWithParameterizedConstructor() {
+ Object instance =
CelebornUtils.createInstance(ParameterizedOnly.class.getName(), "arg");
+ assertTrue(instance instanceof ParameterizedOnly);
+ assertEquals("arg", ((ParameterizedOnly) instance).value);
+ }
+
+ @Test
+ public void testCreateInstancePicksConstructorMatchingArguments() {
+ Object noArg =
CelebornUtils.createInstance(NoArgAndParameterized.class.getName());
+ assertEquals("no-arg", ((NoArgAndParameterized) noArg).value);
+
+ Object withArg =
CelebornUtils.createInstance(NoArgAndParameterized.class.getName(), "arg");
+ assertEquals("arg", ((NoArgAndParameterized) withArg).value);
+ }
+
+ @Test
+ public void testCreateInstanceAcceptsSubtypeArgument() {
+ Object instance =
CelebornUtils.createInstance(SupertypeParameter.class.getName(), "arg");
+ assertTrue(instance instanceof SupertypeParameter);
+ assertEquals("arg", ((SupertypeParameter) instance).value);
+ }
+
+ @Test
+ public void testCreateInstanceAcceptsNullForReferenceParameter() {
+ Object instance =
+ CelebornUtils.createInstance(SupertypeParameter.class.getName(),
(Object) null);
+ assertTrue(instance instanceof SupertypeParameter);
+ assertNull(((SupertypeParameter) instance).value);
+ }
+
+ @Test
+ public void testCreateInstanceRejectsNullForPrimitiveParameter() {
+ try {
+ CelebornUtils.createInstance(PrimitiveParameter.class.getName(),
(Object) null);
+ fail("Expected RuntimeException since no constructor accepts null for a
primitive");
+ } catch (RuntimeException e) {
+ assertTrue(e.getCause() instanceof NoSuchMethodException);
+ }
+ }
+
+ @Test
+ public void testCreateInstanceThrowsWhenNoConstructorMatches() {
+ try {
+ CelebornUtils.createInstance(PrimitiveParameter.class.getName(),
"not-an-int");
+ fail("Expected RuntimeException since no constructor matches the
argument type");
+ } catch (RuntimeException e) {
+ assertTrue(e.getCause() instanceof NoSuchMethodException);
+ }
+ }
+
+ @Test
+ public void testCreateInstanceReturnsNullWhenClassIsAbsent() {
+
assertNull(CelebornUtils.createInstance("org.apache.celeborn.NonexistentClass"));
+ }
+
+ @Test
+ public void
testFetchCleanFailedShuffleReturnsCleanerAndStopShutsDownItsExecutor()
+ throws Exception {
+ FailedShuffleCleanerTestSupport.assumeSupported();
+
+ CelebornConf conf = hermeticConf();
+ conf.set("celeborn.client.spark.fetch.cleanFailedShuffle", "true");
+ // A mocked LifecycleManager keeps the test hermetic: no RpcEnv, no
ApplicationHeartbeater and
+ // therefore no RPC against an ambient Celeborn master.
+ LifecycleManager lifecycleManager = mock(LifecycleManager.class);
+ when(lifecycleManager.conf()).thenReturn(conf);
+
+ Object cleaner = CelebornUtils.fetchCleanFailedShuffle(lifecycleManager,
conf);
+ ExecutorService cleanerExecutor = null;
+ try {
+ assertNotNull("FailedShuffleCleaner instance should be returned to the
caller", cleaner);
+ assertEquals(CelebornUtils.FAILED_SHUFFLE_CLEANER_NAME,
cleaner.getClass().getName());
+ // The returned instance is the very one wired into the LifecycleManager
callbacks.
+
verify(lifecycleManager).registerValidateCelebornShuffleIdForCleanCallback(any());
+ verify(lifecycleManager).registerUnregisterShuffleCallback(any());
+
+ cleanerExecutor = FailedShuffleCleanerTestSupport.executorOf(cleaner);
+ assertNotNull("FailedShuffleCleaner should own a scheduled executor",
cleanerExecutor);
+ assertFalse(
+ "FailedShuffleCleaner executor should be running before stop",
+ cleanerExecutor.isShutdown());
+ } finally {
+ if (cleaner != null) {
+ CelebornUtils.stopFailedShuffleCleaner(cleaner);
+ }
+ }
+
+ assertTrue(
+ "stopFailedShuffleCleaner should shut down the cleaner executor",
+ cleanerExecutor.isShutdown());
+ assertTrue(
+ "cleaner executor should terminate once the cleaner is stopped",
+ cleanerExecutor.awaitTermination(10, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testFetchCleanFailedShuffleDisabledCreatesNoCleaner() {
+ FailedShuffleCleanerTestSupport.assumeSupported();
+
+ CelebornConf conf = hermeticConf();
+ conf.set("celeborn.client.spark.fetch.cleanFailedShuffle", "false");
+ LifecycleManager lifecycleManager = mock(LifecycleManager.class);
+
+ assertNull(CelebornUtils.fetchCleanFailedShuffle(lifecycleManager, conf));
+ verifyNoInteractions(lifecycleManager);
+ }
+
+ @Test
+ public void testStageRerunDisabledReturnsNull() {
+ assertNull(CelebornUtils.stageRerun(false, null, hermeticConf()));
+ }
+
+ /**
+ * Builds a {@link CelebornConf} that ignores ambient {@code celeborn.*}
settings coming from
+ * system properties, so the assertions below cannot be influenced by the
local environment.
+ */
+ private static CelebornConf hermeticConf() {
+ return new CelebornConf(false);
+ }
+}
diff --git
a/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/FailedShuffleCleanerTestSupport.java
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/FailedShuffleCleanerTestSupport.java
new file mode 100644
index 0000000000..f8eb956985
--- /dev/null
+++
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/FailedShuffleCleanerTestSupport.java
@@ -0,0 +1,57 @@
+/*
+ * 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.spark.shuffle.gluten.celeborn;
+
+import org.apache.celeborn.common.CelebornConf;
+import org.junit.Assume;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.ExecutorService;
+
+/** Shared helpers for the {@code FailedShuffleCleaner} lifecycle assertions.
*/
+final class FailedShuffleCleanerTestSupport {
+
+ private FailedShuffleCleanerTestSupport() {}
+
+ /**
+ * Keeps the cleaner assertions strict on Celeborn clients that ship {@code
FailedShuffleCleaner},
+ * and skips them on older clients where {@link
CelebornUtils#fetchCleanFailedShuffle} is a no-op
+ * by design.
+ */
+ static void assumeSupported() {
+ try {
+ CelebornConf.class.getDeclaredMethod("clientFetchCleanFailedShuffle");
+ Class.forName(CelebornUtils.FAILED_SHUFFLE_CLEANER_NAME);
+ } catch (NoSuchMethodException | ClassNotFoundException e) {
+ Assume.assumeNoException("Celeborn client in use does not support
FailedShuffleCleaner", e);
+ }
+ }
+
+ /**
+ * Returns the executor owned by the cleaner, which is the resource that
leaked before the fix.
+ * The field is looked up by type instead of by name to stay tolerant to
Celeborn renames.
+ */
+ static ExecutorService executorOf(Object cleaner) throws
ReflectiveOperationException {
+ for (Field field : cleaner.getClass().getDeclaredFields()) {
+ if (ExecutorService.class.isAssignableFrom(field.getType())) {
+ field.setAccessible(true);
+ return (ExecutorService) field.get(cleaner);
+ }
+ }
+ throw new AssertionError("No ExecutorService field found on " +
cleaner.getClass().getName());
+ }
+}
diff --git
a/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornColumnarBatchSerializerFactory.java
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornColumnarBatchSerializerFactory.java
new file mode 100644
index 0000000000..ba65f70043
--- /dev/null
+++
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornColumnarBatchSerializerFactory.java
@@ -0,0 +1,32 @@
+/*
+ * 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.spark.shuffle.gluten.celeborn;
+
+/**
+ * Satisfies the {@link CelebornShuffleManager} static initializer, which
requires exactly one
+ * factory on the classpath. The real implementations live in the backend
modules, which this module
+ * does not depend on.
+ */
+public class TestCelebornColumnarBatchSerializerFactory
+ implements CelebornColumnarBatchSerializerFactory {
+
+ @Override
+ public String columnarBatchSerializerClass() {
+ throw new UnsupportedOperationException(
+ "Columnar serializers are out of scope for these tests");
+ }
+}
diff --git
a/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornShuffleWriterFactory.java
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornShuffleWriterFactory.java
new file mode 100644
index 0000000000..8e19880589
--- /dev/null
+++
b/gluten-celeborn/src/test/java/org/apache/spark/shuffle/gluten/celeborn/TestCelebornShuffleWriterFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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.spark.shuffle.gluten.celeborn;
+
+import org.apache.celeborn.client.ShuffleClient;
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.spark.TaskContext;
+import org.apache.spark.shuffle.ShuffleWriteMetricsReporter;
+import org.apache.spark.shuffle.ShuffleWriter;
+import org.apache.spark.shuffle.celeborn.CelebornShuffleHandle;
+
+/**
+ * Satisfies the {@link CelebornShuffleManager} static initializer, which
requires exactly one
+ * factory on the classpath. The real implementations live in the backend
modules, which this module
+ * does not depend on.
+ */
+public class TestCelebornShuffleWriterFactory implements
CelebornShuffleWriterFactory {
+
+ @Override
+ public <K, V> ShuffleWriter<K, V> createShuffleWriterInstance(
+ int shuffleId,
+ CelebornShuffleHandle<K, V, V> handle,
+ TaskContext context,
+ CelebornConf celebornConf,
+ ShuffleClient client,
+ ShuffleWriteMetricsReporter writeMetrics) {
+ throw new UnsupportedOperationException("Shuffle writers are out of scope
for these tests");
+ }
+}
diff --git
a/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory
b/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory
new file mode 100644
index 0000000000..6babefbaaa
--- /dev/null
+++
b/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornColumnarBatchSerializerFactory
@@ -0,0 +1 @@
+org.apache.spark.shuffle.gluten.celeborn.TestCelebornColumnarBatchSerializerFactory
diff --git
a/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory
b/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory
new file mode 100644
index 0000000000..aec860240c
--- /dev/null
+++
b/gluten-celeborn/src/test/resources/META-INF/services/org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleWriterFactory
@@ -0,0 +1 @@
+org.apache.spark.shuffle.gluten.celeborn.TestCelebornShuffleWriterFactory
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
