You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geaflow.apache.org by lt...@apache.org on 2026/08/03 12:13:08 UTC
(geaflow) branch master updated: test(cluster): avoid uncaught system exit exceptions (#821)
This is an automated email from the ASF dual-hosted git repository.
Leomrlin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/geaflow.git
The following commit(s) were added to refs/heads/master by this push:
new 6b79587f4 test(cluster): avoid uncaught system exit exceptions (#821)
6b79587f4 is described below
commit 6b79587f48eecc7eeda484c5dde73df42d20f962
Author: Power John <jo...@gmail.com>
AuthorDate: Mon Aug 3 20:13:04 2026 +0800
test(cluster): avoid uncaught system exit exceptions (#821)
---
.../exception/ComponentExceptionSupervisor.java | 20 ++++++-
.../geaflow/cluster/container/ContainerTest.java | 68 ++++++++++++----------
.../ComponentUncaughtExceptionHandlerTest.java | 58 ++++++++----------
.../cluster/util/SystemExitSignalCatcher.java | 48 ---------------
4 files changed, 82 insertions(+), 112 deletions(-)
diff --git a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/org/apache/geaflow/cluster/exception/ComponentExceptionSupervisor.java b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/org/apache/geaflow/cluster/exception/ComponentExceptionSupervisor.java
index 145174b2b..13a8cec25 100644
--- a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/org/apache/geaflow/cluster/exception/ComponentExceptionSupervisor.java
+++ b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/org/apache/geaflow/cluster/exception/ComponentExceptionSupervisor.java
@@ -21,6 +21,9 @@ package org.apache.geaflow.cluster.exception;
import static org.apache.geaflow.cluster.constants.ClusterConstants.EXIT_CODE;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Objects;
+import java.util.function.IntConsumer;
import org.apache.geaflow.cluster.task.runner.AbstractTaskRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,6 +33,16 @@ public class ComponentExceptionSupervisor extends AbstractTaskRunner<ComponentEx
private static final Logger LOGGER = LoggerFactory.getLogger(ComponentExceptionSupervisor.class);
private static ComponentExceptionSupervisor INSTANCE;
+ private final IntConsumer processExit;
+
+ public ComponentExceptionSupervisor() {
+ this(System::exit);
+ }
+
+ @VisibleForTesting
+ protected ComponentExceptionSupervisor(IntConsumer processExit) {
+ this.processExit = Objects.requireNonNull(processExit);
+ }
@Override
protected void process(ExceptionElement exceptionElement) {
@@ -42,7 +55,7 @@ public class ComponentExceptionSupervisor extends AbstractTaskRunner<ComponentEx
if (running) {
LOGGER.error(String.format("%s occur fatal exception, exit process now",
exceptionElement.thread), exceptionElement.cause);
- System.exit(EXIT_CODE);
+ processExit.accept(EXIT_CODE);
} else {
LOGGER.info("{} ignore exception because supervisor is shutdown", exceptionElement.thread);
}
@@ -55,6 +68,11 @@ public class ComponentExceptionSupervisor extends AbstractTaskRunner<ComponentEx
return INSTANCE;
}
+ @VisibleForTesting
+ protected static synchronized void setInstance(ComponentExceptionSupervisor instance) {
+ INSTANCE = Objects.requireNonNull(instance);
+ }
+
@Override
public void shutdown() {
super.shutdown();
diff --git a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/container/ContainerTest.java b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/container/ContainerTest.java
index afb457056..5945d953c 100644
--- a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/container/ContainerTest.java
+++ b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/container/ContainerTest.java
@@ -23,47 +23,38 @@ import static org.apache.geaflow.common.config.keys.ExecutionConfigKeys.CLUSTER_
import static org.apache.geaflow.common.config.keys.ExecutionConfigKeys.CONTAINER_DISPATCH_THREADS;
import static org.apache.geaflow.common.config.keys.ExecutionConfigKeys.REPORTER_LIST;
import static org.apache.geaflow.common.config.keys.ExecutionConfigKeys.RUN_LOCAL_MODE;
+import static org.apache.geaflow.cluster.constants.ClusterConstants.EXIT_CODE;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntConsumer;
+import org.apache.geaflow.cluster.exception.ComponentExceptionSupervisor;
import org.apache.geaflow.cluster.exception.ExceptionCollectService;
import org.apache.geaflow.cluster.protocol.EventType;
import org.apache.geaflow.cluster.protocol.ICommand;
import org.apache.geaflow.cluster.protocol.IExecutableCommand;
import org.apache.geaflow.cluster.protocol.OpenContainerEvent;
import org.apache.geaflow.cluster.task.ITaskContext;
-import org.apache.geaflow.cluster.util.SystemExitSignalCatcher;
import org.apache.geaflow.common.config.Configuration;
import org.apache.geaflow.common.utils.ReflectionUtil;
import org.apache.geaflow.common.utils.SleepUtils;
import org.apache.geaflow.ha.service.HAServiceFactory;
import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
public class ContainerTest {
+ private static final int EXIT_NOT_CALLED = Integer.MIN_VALUE;
private static AtomicBoolean eventExecuted = new AtomicBoolean(false);
- private static AtomicBoolean hasException = new AtomicBoolean(false);
- private static SecurityManager securityManager;
-
- @BeforeClass
- public void before() {
- securityManager = System.getSecurityManager();
- System.setSecurityManager(new SystemExitSignalCatcher(hasException));
- }
-
- @AfterClass
- public void after() {
- System.setSecurityManager(securityManager);
- }
+ private static AtomicInteger exitCode = new AtomicInteger(EXIT_NOT_CALLED);
@BeforeMethod
public void beforeMethod() {
- hasException.set(false);
+ exitCode.set(EXIT_NOT_CALLED);
+ TestComponentExceptionSupervisor.install(exitCode::set);
}
@Test
@@ -83,12 +74,15 @@ public class ContainerTest {
ReflectionUtil.setField(container, "containerContext", new ContainerContext(0, configuration));
ReflectionUtil.setField(container, "exceptionCollectService", new ExceptionCollectService());
container.open(new OpenContainerEvent(1));
- container.process(new TestCreateTaskEvent());
- container.process(new ExceptionCommandEvent());
-
- waitTestResult();
- Assert.assertTrue(hasException.get());
- container.close();
+ try {
+ container.process(new TestCreateTaskEvent());
+ container.process(new ExceptionCommandEvent());
+
+ waitTestResult();
+ Assert.assertEquals(exitCode.get(), EXIT_CODE);
+ } finally {
+ container.close();
+ }
}
@Test
@@ -108,12 +102,15 @@ public class ContainerTest {
ReflectionUtil.setField(container, "containerContext", new ContainerContext(0, configuration));
ReflectionUtil.setField(container, "exceptionCollectService", new ExceptionCollectService());
container.open(new OpenContainerEvent(1));
- container.process(new TestCreateTaskEvent());
- container.process(new ExceptionCommandEvent());
-
- waitTestResult();
- Assert.assertTrue(hasException.get());
- container.close();
+ try {
+ container.process(new TestCreateTaskEvent());
+ container.process(new ExceptionCommandEvent());
+
+ waitTestResult();
+ Assert.assertEquals(exitCode.get(), EXIT_CODE);
+ } finally {
+ container.close();
+ }
}
private void waitTestResult() {
@@ -123,7 +120,7 @@ public class ContainerTest {
retry--;
}
retry = 10;
- while (!hasException.get() && retry > 0) {
+ while (exitCode.get() == EXIT_NOT_CALLED && retry > 0) {
SleepUtils.sleepMilliSecond(100);
retry--;
}
@@ -165,4 +162,15 @@ public class ContainerTest {
return EventType.CREATE_TASK;
}
}
+
+ private static class TestComponentExceptionSupervisor extends ComponentExceptionSupervisor {
+
+ private TestComponentExceptionSupervisor(IntConsumer processExit) {
+ super(processExit);
+ }
+
+ private static void install(IntConsumer processExit) {
+ setInstance(new TestComponentExceptionSupervisor(processExit));
+ }
+ }
}
diff --git a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/exception/ComponentUncaughtExceptionHandlerTest.java b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/exception/ComponentUncaughtExceptionHandlerTest.java
index 1bae92aca..38426508f 100644
--- a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/exception/ComponentUncaughtExceptionHandlerTest.java
+++ b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/exception/ComponentUncaughtExceptionHandlerTest.java
@@ -19,52 +19,44 @@
package org.apache.geaflow.cluster.exception;
+import static org.apache.geaflow.cluster.constants.ClusterConstants.EXIT_CODE;
+
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.geaflow.cluster.util.SystemExitSignalCatcher;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.geaflow.common.utils.ThreadUtil;
import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
-import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
public class ComponentUncaughtExceptionHandlerTest {
- private static SecurityManager securityManager;
- private static AtomicBoolean hasException = new AtomicBoolean(false);
-
-
- @BeforeClass
- public void before() {
- securityManager = System.getSecurityManager();
- System.setSecurityManager(new SystemExitSignalCatcher(hasException));
- }
-
- @AfterClass
- public void after() {
- System.setSecurityManager(securityManager);
- }
-
- @BeforeMethod
- public void beforeMethod() {
- hasException.set(false);
- }
-
@Test
public void testHandleExceptionInThreadPool() throws InterruptedException {
- ComponentExceptionSupervisor.getInstance();
+ AtomicInteger exitCode = new AtomicInteger();
+ CountDownLatch exitCalled = new CountDownLatch(1);
+ ComponentExceptionSupervisor supervisor = new ComponentExceptionSupervisor(code -> {
+ exitCode.set(code);
+ exitCalled.countDown();
+ });
+ ComponentExceptionSupervisor.setInstance(supervisor);
ExecutorService executorService = Executors.newFixedThreadPool(2,
ThreadUtil.namedThreadFactory(true, "test-handler", new ComponentUncaughtExceptionHandler()));
- executorService.execute(() -> {
- throw new RuntimeException("test exception");
- });
- executorService.execute(ComponentExceptionSupervisor.getInstance());
- // wait async thread catch and handle exception
- Thread.sleep(100);
- Assert.assertTrue(hasException.get());
+ try {
+ executorService.execute(() -> {
+ throw new RuntimeException("test exception");
+ });
+ executorService.execute(supervisor);
+
+ Assert.assertTrue(exitCalled.await(1, TimeUnit.SECONDS));
+ Assert.assertEquals(exitCode.get(), EXIT_CODE);
+ } finally {
+ supervisor.shutdown();
+ executorService.shutdown();
+ executorService.awaitTermination(1, TimeUnit.SECONDS);
+ }
}
}
diff --git a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/util/SystemExitSignalCatcher.java b/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/util/SystemExitSignalCatcher.java
deleted file mode 100644
index bf9c9ee2b..000000000
--- a/geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/util/SystemExitSignalCatcher.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.geaflow.cluster.util;
-
-import java.security.Permission;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.geaflow.common.exception.GeaflowRuntimeException;
-
-public class SystemExitSignalCatcher extends SecurityManager {
-
- private AtomicBoolean hasSignal;
-
- public SystemExitSignalCatcher(AtomicBoolean hasSignal) {
- this.hasSignal = hasSignal;
- }
-
- @Override
- public void checkPermission(Permission perm) {
- }
-
- @Override
- public void checkPermission(Permission perm, Object context) {
- }
-
- @Override
- public void checkExit(int status) {
- super.checkExit(status);
- hasSignal.set(true);
- throw new GeaflowRuntimeException("throw exception instead of exit process");
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@geaflow.apache.org
For additional commands, e-mail: commits-help@geaflow.apache.org