This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 811702191a [Improve][Zeta] Refactor coordinator executor creation and
fix failover lifecycle issues (#10865)
811702191a is described below
commit 811702191a7d87236a68cc285a03e868d8eece67
Author: zhiwei.niu <[email protected]>
AuthorDate: Thu May 14 22:37:59 2026 +0800
[Improve][Zeta] Refactor coordinator executor creation and fix failover
lifecycle issues (#10865)
---
.../SplitClusterPendingJobLifecycleFailoverIT.java | 396 +++++++++++++++++++++
.../pending_jobs_streaming_lifecycle.conf | 42 +++
.../engine/server/CoordinatorService.java | 41 ++-
.../engine/server/CoordinatorServiceTest.java | 286 ++++++++++++++-
4 files changed, 736 insertions(+), 29 deletions(-)
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
new file mode 100644
index 0000000000..17033d47b4
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/SplitClusterPendingJobLifecycleFailoverIT.java
@@ -0,0 +1,396 @@
+/*
+ * 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.seatunnel.engine.e2e;
+
+import org.apache.seatunnel.common.config.Common;
+import org.apache.seatunnel.common.config.DeployMode;
+import org.apache.seatunnel.engine.client.SeaTunnelClient;
+import org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment;
+import org.apache.seatunnel.engine.client.job.ClientJobProxy;
+import org.apache.seatunnel.engine.common.config.ConfigProvider;
+import org.apache.seatunnel.engine.common.config.JobConfig;
+import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.common.config.server.ScheduleStrategy;
+import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
+import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.SeaTunnelServerStarter;
+
+import org.awaitility.Awaitility;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.instance.impl.HazelcastInstanceImpl;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public class SplitClusterPendingJobLifecycleFailoverIT {
+ private static final String JOB_CONFIG_FILE =
"pending_jobs_streaming_lifecycle.conf";
+
+ @Test
+ public void testPendingJobLifecycleInMasterFailover() {
+ String testClusterName =
+
"SplitClusterPendingJobLifecycleFailoverIT_testPendingJobLifecycleInMasterFailover";
+ HazelcastInstanceImpl masterNode1 = null;
+ HazelcastInstanceImpl masterNode2 = null;
+ HazelcastInstanceImpl workerNode1 = null;
+ HazelcastInstanceImpl workerNode2 = null;
+ SeaTunnelClient engineClient = null;
+ ClientJobProxy holderJob = null;
+ ClientJobProxy pendingJob = null;
+
+ SeaTunnelConfig masterNode1Config =
getSeaTunnelConfig(testClusterName);
+ SeaTunnelConfig masterNode2Config =
getSeaTunnelConfig(testClusterName);
+ SeaTunnelConfig workerNode1Config =
getSeaTunnelConfig(testClusterName);
+ SeaTunnelConfig workerNode2Config =
getSeaTunnelConfig(testClusterName);
+ configurePendingLifecycleTest(masterNode1Config);
+ configurePendingLifecycleTest(masterNode2Config);
+ configurePendingLifecycleTest(workerNode1Config);
+ configurePendingLifecycleTest(workerNode2Config);
+
+ try {
+ masterNode1 =
SeaTunnelServerStarter.createMasterHazelcastInstance(masterNode1Config);
+ masterNode2 =
SeaTunnelServerStarter.createMasterHazelcastInstance(masterNode2Config);
+ workerNode1 =
SeaTunnelServerStarter.createWorkerHazelcastInstance(workerNode1Config);
+
+ HazelcastInstanceImpl finalMasterNode = masterNode1;
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 3,
finalMasterNode.getCluster().getMembers().size()));
+
+ Common.setDeployMode(DeployMode.CLUSTER);
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+
clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+
+ holderJob =
+ submitJob(
+ engineClient,
+ masterNode1Config,
+ "pending_job_lifecycle_holder",
+ TestUtils.getResource(JOB_CONFIG_FILE));
+ assertJobStatusWithTimeout(holderJob, JobStatus.RUNNING, 120);
+
+ HazelcastInstanceImpl activeMaster =
waitAndFindActiveMaster(masterNode1, masterNode2);
+ assertPendingQueueState(activeMaster, null, 0);
+
+ pendingJob =
+ submitJob(
+ engineClient,
+ masterNode1Config,
+ "pending_job_lifecycle_pending",
+ TestUtils.getResource(JOB_CONFIG_FILE));
+ final ClientJobProxy finalPendingJob = pendingJob;
+ final long pendingJobId = finalPendingJob.getJobId();
+ assertJobStatusWithTimeout(pendingJob, JobStatus.PENDING, 120);
+ assertPendingQueueState(activeMaster, pendingJobId, 1);
+ activeMaster.shutdown();
+ HazelcastInstanceImpl standbyMaster =
+ activeMaster == masterNode1 ? masterNode2 : masterNode1;
+
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+
standbyMaster.getLifecycleService().isRunning());
+ Assertions.assertEquals(
+ 2,
standbyMaster.getCluster().getMembers().size());
+ });
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+ isCoordinatorActive(standbyMaster),
+ "Standby master should become
active after failover"));
+ ClientJobProxy pendingJobAfterFailover =
+ engineClient.createJobClient().getJobProxy(pendingJobId);
+ assertPendingQueueContainsJob(standbyMaster, pendingJobId, 1);
+
+ Awaitility.await()
+ .during(10, TimeUnit.SECONDS)
+ .atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertPendingQueueContainsJob(standbyMaster,
pendingJobId, 1);
+ Assertions.assertEquals(
+ JobStatus.PENDING,
pendingJobAfterFailover.getJobStatus());
+ });
+
+ workerNode2 =
SeaTunnelServerStarter.createWorkerHazelcastInstance(workerNode2Config);
+ Awaitility.await()
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 3,
standbyMaster.getCluster().getMembers().size()));
+ assertJobStatusWithTimeout(pendingJobAfterFailover,
JobStatus.RUNNING, 180);
+ assertPendingQueueNotContainsJob(standbyMaster, pendingJobId);
+
+ pendingJobAfterFailover.cancelJob();
+ assertJobStatusWithTimeout(pendingJobAfterFailover,
JobStatus.CANCELED, 120);
+
engineClient.createJobClient().getJobProxy(holderJob.getJobId()).cancelJob();
+ } finally {
+ if (engineClient != null) {
+ engineClient.close();
+ }
+ if (masterNode1 != null) {
+ masterNode1.shutdown();
+ }
+ if (masterNode2 != null) {
+ masterNode2.shutdown();
+ }
+ if (workerNode1 != null) {
+ workerNode1.shutdown();
+ }
+ if (workerNode2 != null) {
+ workerNode2.shutdown();
+ }
+ }
+ }
+
+ @Test
+ public void testPendingJobScheduledAfterRunningJobCanceled() {
+ String testClusterName =
+
"SplitClusterPendingJobLifecycleFailoverIT_testPendingJobScheduledAfterRunningJobCanceled";
+ HazelcastInstanceImpl masterNode = null;
+ HazelcastInstanceImpl workerNode = null;
+ SeaTunnelClient engineClient = null;
+ ClientJobProxy holderJob = null;
+ ClientJobProxy pendingJob = null;
+
+ SeaTunnelConfig masterNodeConfig = getSeaTunnelConfig(testClusterName);
+ SeaTunnelConfig workerNodeConfig = getSeaTunnelConfig(testClusterName);
+ configurePendingLifecycleTest(masterNodeConfig);
+ configurePendingLifecycleTest(workerNodeConfig);
+
+ try {
+ masterNode =
SeaTunnelServerStarter.createMasterHazelcastInstance(masterNodeConfig);
+ workerNode =
SeaTunnelServerStarter.createWorkerHazelcastInstance(workerNodeConfig);
+
+ HazelcastInstanceImpl finalMasterNode = masterNode;
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ 2,
finalMasterNode.getCluster().getMembers().size()));
+
+ Common.setDeployMode(DeployMode.CLUSTER);
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+
clientConfig.setClusterName(TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+
+ holderJob =
+ submitJob(
+ engineClient,
+ masterNodeConfig,
+ "pending_job_lifecycle_holder_cancel",
+ TestUtils.getResource(JOB_CONFIG_FILE));
+ assertJobStatusWithTimeout(holderJob, JobStatus.RUNNING, 120);
+
+ HazelcastInstanceImpl activeMaster =
waitAndFindActiveMaster(masterNode, null);
+ assertPendingQueueState(activeMaster, null, 0);
+
+ pendingJob =
+ submitJob(
+ engineClient,
+ masterNodeConfig,
+ "pending_job_lifecycle_pending_after_cancel",
+ TestUtils.getResource(JOB_CONFIG_FILE));
+ long pendingJobId = pendingJob.getJobId();
+ assertJobStatusWithTimeout(pendingJob, JobStatus.PENDING, 120);
+ assertPendingQueueState(activeMaster, pendingJobId, 1);
+
+ holderJob.cancelJob();
+ assertJobStatusWithTimeout(holderJob, JobStatus.CANCELED, 120);
+
+ assertJobStatusWithTimeout(pendingJob, JobStatus.RUNNING, 180);
+ assertPendingQueueNotContainsJob(activeMaster, pendingJobId);
+
+ pendingJob.cancelJob();
+ assertJobStatusWithTimeout(pendingJob, JobStatus.CANCELED, 120);
+ } finally {
+ if (engineClient != null) {
+ engineClient.close();
+ }
+ if (masterNode != null) {
+ masterNode.shutdown();
+ }
+ if (workerNode != null) {
+ workerNode.shutdown();
+ }
+ }
+ }
+
+ @NotNull private static SeaTunnelConfig getSeaTunnelConfig(String
testClusterName) {
+ SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
+ seaTunnelConfig
+ .getHazelcastConfig()
+ .setClusterName(TestUtils.getClusterName(testClusterName));
+ return seaTunnelConfig;
+ }
+
+ private static void configurePendingLifecycleTest(SeaTunnelConfig
seaTunnelConfig) {
+
seaTunnelConfig.getEngineConfig().getSlotServiceConfig().setDynamicSlot(false);
+ seaTunnelConfig.getEngineConfig().getSlotServiceConfig().setSlotNum(4);
+
seaTunnelConfig.getEngineConfig().setScheduleStrategy(ScheduleStrategy.WAIT);
+ seaTunnelConfig.getEngineConfig().getHttpConfig().setEnabled(false);
+ }
+
+ private static ClientJobProxy submitJob(
+ SeaTunnelClient engineClient,
+ SeaTunnelConfig seaTunnelConfig,
+ String jobName,
+ String jobConfigFile) {
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName(jobName);
+ ClientJobExecutionEnvironment jobExecutionEnv =
+ engineClient.createExecutionContext(jobConfigFile, jobConfig,
seaTunnelConfig);
+ try {
+ return jobExecutionEnv.execute();
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Failed to submit job " + jobName, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted when submitting job " +
jobName, e);
+ }
+ }
+
+ private static void assertJobStatusWithTimeout(
+ ClientJobProxy clientJobProxy, JobStatus expectedStatus, long
timeoutSeconds) {
+ Awaitility.await()
+ .atMost(timeoutSeconds, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ expectedStatus,
clientJobProxy.getJobStatus()));
+ }
+
+ private static HazelcastInstanceImpl waitAndFindActiveMaster(
+ HazelcastInstanceImpl masterNode1, HazelcastInstanceImpl
masterNode2) {
+ final HazelcastInstanceImpl[] activeMasterRef = new
HazelcastInstanceImpl[1];
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ activeMasterRef[0] = findActiveMaster(masterNode1,
masterNode2);
+ Assertions.assertNotNull(
+ activeMasterRef[0],
+ "Should find active master after
coordinator initialization");
+ });
+ return activeMasterRef[0];
+ }
+
+ private static HazelcastInstanceImpl findActiveMaster(
+ HazelcastInstanceImpl masterNode1, HazelcastInstanceImpl
masterNode2) {
+ if (isCoordinatorActive(masterNode1)) {
+ return masterNode1;
+ }
+ if (isCoordinatorActive(masterNode2)) {
+ return masterNode2;
+ }
+ return null;
+ }
+
+ private static boolean isCoordinatorActive(HazelcastInstanceImpl
masterNode) {
+ if (masterNode == null ||
!masterNode.getLifecycleService().isRunning()) {
+ return false;
+ }
+ SeaTunnelServer server =
+
masterNode.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ try {
+ return server.getCoordinatorService().isCoordinatorActive();
+ } catch (SeaTunnelEngineException e) {
+ return false;
+ }
+ }
+
+ private static void assertPendingQueueState(
+ HazelcastInstanceImpl masterNode, Long expectedJobIdInQueue, int
expectedPendingCount) {
+ SeaTunnelServer server =
+
masterNode.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+
server.getCoordinatorService().isCoordinatorActive(),
+ "Coordinator should be active when
asserting pending queue");
+ Assertions.assertEquals(
+ expectedPendingCount,
+
server.getCoordinatorService().getPendingJobCount());
+ if (expectedJobIdInQueue != null) {
+ Assertions.assertTrue(
+ server.getCoordinatorService()
+ .getPendingJobQueue()
+
.contains(expectedJobIdInQueue),
+ "Expected pending job should remain in
pending queue");
+ }
+ });
+ }
+
+ private static void assertPendingQueueContainsJob(
+ HazelcastInstanceImpl masterNode,
+ long expectedJobIdInQueue,
+ int minExpectedPendingCount) {
+ SeaTunnelServer server =
+
masterNode.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+
server.getCoordinatorService().isCoordinatorActive(),
+ "Coordinator should be active when
asserting pending queue");
+ Assertions.assertTrue(
+
server.getCoordinatorService().getPendingJobCount()
+ >= minExpectedPendingCount,
+ "Pending queue count should be at least "
+ + minExpectedPendingCount);
+ Assertions.assertTrue(
+ server.getCoordinatorService()
+ .getPendingJobQueue()
+ .contains(expectedJobIdInQueue),
+ "Expected pending job should remain in
pending queue");
+ });
+ }
+
+ private static void assertPendingQueueNotContainsJob(
+ HazelcastInstanceImpl masterNode, long expectedRemovedJobId) {
+ SeaTunnelServer server =
+
masterNode.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME);
+ Awaitility.await()
+ .atMost(20, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertFalse(
+ server.getCoordinatorService()
+ .getPendingJobQueue()
+
.contains(expectedRemovedJobId),
+ "Pending queue should not contain job "
+ + expectedRemovedJobId));
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/pending_jobs_streaming_lifecycle.conf
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/pending_jobs_streaming_lifecycle.conf
new file mode 100644
index 0000000000..4317278dd2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/pending_jobs_streaming_lifecycle.conf
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+
+env {
+ job.mode = "STREAMING"
+}
+
+source {
+ FakeSource {
+ parallelism = 2
+ row.num = 1000000
+ split.read-interval = 100
+ schema = {
+ fields {
+ c_int = int
+ }
+ }
+ }
+}
+
+transform {
+}
+
+sink {
+ InMemory {
+ writer_sleep = true
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
index fffe655cf8..a79ff5f2f1 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
@@ -223,17 +223,7 @@ public class CoordinatorService {
this.nodeEngine = nodeEngine;
this.engineConfig = engineConfig;
this.logger = nodeEngine.getLogger(getClass());
- this.executorService =
- new ThreadPoolExecutor(
-
engineConfig.getCoordinatorServiceConfig().getCoreThreadNum(),
-
engineConfig.getCoordinatorServiceConfig().getMaxThreadNum(),
- 60L,
- TimeUnit.SECONDS,
- new SynchronousQueue<>(),
- new ThreadFactoryBuilder()
-
.setNameFormat("seatunnel-coordinator-service-%d")
- .build(),
- new ThreadPoolStatus.RejectionCountingHandler());
+ this.executorService = createCoordinatorExecutor();
this.seaTunnelServer = seaTunnelServer;
masterActiveListener = Executors.newSingleThreadScheduledExecutor();
@@ -251,9 +241,19 @@ public class CoordinatorService {
TimeUnit.SECONDS);
scheduleStrategy = engineConfig.getScheduleStrategy();
isWaitStrategy = scheduleStrategy.equals(ScheduleStrategy.WAIT);
- logger.info("Start pending job schedule thread");
- // start pending job schedule thread
- startPendingJobScheduleThread();
+ }
+
+ private ExecutorService createCoordinatorExecutor() {
+ return new ThreadPoolExecutor(
+ engineConfig.getCoordinatorServiceConfig().getCoreThreadNum(),
+ engineConfig.getCoordinatorServiceConfig().getMaxThreadNum(),
+ 60L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new ThreadFactoryBuilder()
+ .setNameFormat("seatunnel-coordinator-service-%d")
+ .build(),
+ new ThreadPoolStatus.RejectionCountingHandler());
}
/**
@@ -264,10 +264,11 @@ public class CoordinatorService {
* does not terminate future pending-job processing.
*/
private void startPendingJobScheduleThread() {
+ logger.info("Start pending job schedule thread");
Runnable pendingJobScheduleTask =
() -> {
Thread.currentThread().setName("pending-job-schedule-runner");
- while (true) {
+ while (isActive) {
try {
pendingJobSchedule();
} catch (InterruptedException interrupted) {
@@ -1010,15 +1011,12 @@ public class CoordinatorService {
if (!isActive && this.seaTunnelServer.isMasterNode()) {
logger.info(
"This node become a new active master node, begin init
coordinator service");
- if (this.executorService.isShutdown()) {
- this.executorService =
- Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
-
.setNameFormat("seatunnel-coordinator-service-%d")
- .build());
+ if (this.executorService.isShutdown() ||
this.executorService.isTerminated()) {
+ this.executorService = createCoordinatorExecutor();
}
initCoordinatorService();
isActive = true;
+ startPendingJobScheduleThread();
} else if (isActive && !this.seaTunnelServer.isMasterNode()) {
isActive = false;
logger.info(
@@ -1555,6 +1553,7 @@ public class CoordinatorService {
}
public void shutdown() {
+ isActive = false;
if (masterActiveListener != null) {
masterActiveListener.shutdownNow();
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
index 195e0b878a..484546676c 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
@@ -20,18 +20,25 @@ package org.apache.seatunnel.engine.server;
import org.apache.seatunnel.common.utils.ReflectionUtils;
import org.apache.seatunnel.engine.common.Constant;
import org.apache.seatunnel.engine.common.config.ConfigProvider;
+import org.apache.seatunnel.engine.common.config.EngineConfig;
import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
+import org.apache.seatunnel.engine.common.config.server.ScheduleStrategy;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
+import org.apache.seatunnel.engine.common.job.JobResult;
import org.apache.seatunnel.engine.common.job.JobStatus;
+import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture;
import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import org.apache.seatunnel.engine.core.dag.logical.LogicalDag;
import org.apache.seatunnel.engine.core.job.JobDAGInfo;
import org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.JobInfo;
import org.apache.seatunnel.engine.core.job.PipelineStatus;
+import org.apache.seatunnel.engine.server.dag.physical.PhysicalPlan;
import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.dag.physical.SubPlan;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
+import org.apache.seatunnel.engine.server.execution.PendingJobInfo;
+import org.apache.seatunnel.engine.server.execution.PendingSourceState;
import org.apache.seatunnel.engine.server.execution.TaskGroupContext;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.execution.TaskLocation;
@@ -51,6 +58,7 @@ import org.mockito.Mockito;
import com.hazelcast.instance.impl.HazelcastInstanceImpl;
import com.hazelcast.internal.serialization.Data;
+import com.hazelcast.logging.ILogger;
import com.hazelcast.map.IMap;
import com.hazelcast.spi.exception.RetryableHazelcastException;
import com.hazelcast.spi.impl.NodeEngineImpl;
@@ -70,7 +78,10 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static
org.apache.seatunnel.engine.core.classloader.DefaultClassLoaderService.SKIP_CHECK_JAR;
import static org.awaitility.Awaitility.await;
@@ -157,10 +168,13 @@ public class CoordinatorServiceTest {
await().atMost(120, TimeUnit.SECONDS)
.untilAsserted(
- () ->
- Assertions.assertEquals(
- JobStatus.CANCELED,
-
coordinatorService.getJobStatus(jobId)));
+ () -> {
+ JobStatus status =
coordinatorService.getJobStatus(jobId);
+ Assertions.assertTrue(
+ status == JobStatus.CANCELING || status ==
JobStatus.CANCELED,
+ "Expected job status to be CANCELING or
CANCELED, but got "
+ + status);
+ });
// Ensure the real execution is gone before injecting stale zombie
metadata.
await().atMost(30, TimeUnit.SECONDS)
@@ -258,7 +272,7 @@ public class CoordinatorServiceTest {
return false;
}
} catch (Exception e) {
- return true;
+ continue;
}
}
return true;
@@ -310,6 +324,175 @@ public class CoordinatorServiceTest {
instance2.shutdown();
}
+ @Test
+ void testCheckNewActiveMasterCanSchedulePendingQueue() throws Exception {
+ AtomicBoolean masterFlag = new AtomicBoolean(false);
+ SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(server.isMasterNode()).thenAnswer(invocation ->
masterFlag.get());
+ CoordinatorService coordinatorService =
newMockCoordinatorService(server);
+ try {
+ CountDownLatch runLatch = new CountDownLatch(1);
+ JobMaster jobMaster = enqueueMockPendingJob(coordinatorService,
10001L, runLatch);
+
+ masterFlag.set(true);
+ invokeCheckNewActiveMaster(coordinatorService);
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(0L,
runLatch.getCount()));
+ Mockito.verify(jobMaster, Mockito.times(1)).run();
+ } finally {
+ coordinatorService.shutdown();
+ }
+ }
+
+ @Test
+ void testFailoverStopsOldPendingQueueAndNewCoordinatorCanSchedule() throws
Exception {
+ AtomicBoolean oldMasterFlag = new AtomicBoolean(false);
+ SeaTunnelServer oldServer = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(oldServer.isMasterNode()).thenAnswer(invocation ->
oldMasterFlag.get());
+ CoordinatorService oldCoordinator =
newMockCoordinatorService(oldServer);
+ try {
+ oldMasterFlag.set(true);
+ invokeCheckNewActiveMaster(oldCoordinator);
+
+ await().atMost(15, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(oldCoordinator.isCoordinatorActive()));
+
+ oldMasterFlag.set(false);
+ invokeCheckNewActiveMaster(oldCoordinator);
+
Assertions.assertTrue(getCoordinatorExecutor(oldCoordinator).isShutdown());
+
+ CountDownLatch oldRunLatch = new CountDownLatch(1);
+ JobMaster oldPendingJob = enqueueMockPendingJob(oldCoordinator,
20001L, oldRunLatch);
+
Assertions.assertTrue(oldCoordinator.getPendingJobQueue().contains(20001L));
+ await().during(1, TimeUnit.SECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(1L,
oldRunLatch.getCount()));
+ Mockito.verify(oldPendingJob, Mockito.never()).run();
+ Assertions.assertTrue(
+ oldCoordinator.getPendingJobQueue().contains(20001L),
+ "old coordinator pending queue should not be consumed
after failover");
+
+ AtomicBoolean newMasterFlag = new AtomicBoolean(true);
+ SeaTunnelServer newServer = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(newServer.isMasterNode()).thenAnswer(invocation ->
newMasterFlag.get());
+ CoordinatorService newCoordinator =
newMockCoordinatorService(newServer);
+ try {
+ CountDownLatch newRunLatch = new CountDownLatch(1);
+ JobMaster newPendingJob =
+ enqueueMockPendingJob(newCoordinator, 30001L,
newRunLatch);
+
+ invokeCheckNewActiveMaster(newCoordinator);
+
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(newCoordinator.isCoordinatorActive()));
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(0L,
newRunLatch.getCount()));
+ Mockito.verify(newPendingJob, Mockito.times(1)).run();
+
Assertions.assertFalse(newCoordinator.getPendingJobQueue().contains(30001L));
+ } finally {
+ newCoordinator.shutdown();
+ }
+ } finally {
+ if (!getCoordinatorExecutor(oldCoordinator).isShutdown()) {
+ oldCoordinator.shutdown();
+ }
+ }
+ }
+
+ @Test
+ void testCheckNewActiveMasterIsIdempotentWhenAlreadyActive() throws
Exception {
+ AtomicBoolean masterFlag = new AtomicBoolean(true);
+ SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(server.isMasterNode()).thenAnswer(invocation ->
masterFlag.get());
+ CoordinatorService coordinatorService =
newMockCoordinatorService(server);
+ try {
+ CountDownLatch runLatch = new CountDownLatch(1);
+ JobMaster jobMaster = enqueueMockPendingJob(coordinatorService,
40001L, runLatch);
+
+ invokeCheckNewActiveMaster(coordinatorService);
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(0L,
runLatch.getCount()));
+
+ invokeCheckNewActiveMaster(coordinatorService);
+ await().during(1, TimeUnit.SECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> Mockito.verify(jobMaster,
Mockito.times(1)).run());
+ } finally {
+ coordinatorService.shutdown();
+ }
+ }
+
+ @Test
+ void testPendingJobWithInsufficientResourceRespectsWaitStrategy() throws
Exception {
+ AtomicBoolean masterFlag = new AtomicBoolean(true);
+ SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(server.isMasterNode()).thenAnswer(invocation ->
masterFlag.get());
+
+ EngineConfig engineConfig = new EngineConfig();
+ engineConfig.setScheduleStrategy(ScheduleStrategy.WAIT);
+ CoordinatorService coordinatorService =
newMockCoordinatorService(server, engineConfig);
+ try {
+ CountDownLatch runLatch = new CountDownLatch(1);
+ JobMaster jobMaster =
+ enqueueMockPendingJob(coordinatorService, 50001L,
runLatch, false);
+
+ invokeCheckNewActiveMaster(coordinatorService);
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+ await().during(1, TimeUnit.SECONDS)
+ .atMost(2, TimeUnit.SECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(1L,
runLatch.getCount()));
+
+ Mockito.verify(jobMaster, Mockito.never()).run();
+
Assertions.assertTrue(coordinatorService.getPendingJobQueue().contains(50001L));
+ } finally {
+ coordinatorService.shutdown();
+ }
+ }
+
+ @Test
+ void testPendingJobWithInsufficientResourceRespectsRejectStrategy() throws
Exception {
+ AtomicBoolean masterFlag = new AtomicBoolean(true);
+ SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
+ Mockito.when(server.isMasterNode()).thenAnswer(invocation ->
masterFlag.get());
+
+ EngineConfig engineConfig = new EngineConfig();
+ engineConfig.setScheduleStrategy(ScheduleStrategy.REJECT);
+ CoordinatorService coordinatorService =
newMockCoordinatorService(server, engineConfig);
+ try {
+ CountDownLatch runLatch = new CountDownLatch(1);
+ JobMaster jobMaster =
+ enqueueMockPendingJob(coordinatorService, 60001L,
runLatch, false);
+
+ invokeCheckNewActiveMaster(coordinatorService);
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+ await().atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertFalse(
+ coordinatorService
+ .getPendingJobQueue()
+ .contains(60001L)));
+
+ Mockito.verify(jobMaster, Mockito.never()).run();
+ Assertions.assertEquals(1L, runLatch.getCount());
+ } finally {
+ coordinatorService.shutdown();
+ }
+ }
+
private HazelcastInstanceImpl createHazelcastInstanceWithJoinPortTryCount(
String clusterName, int joinPortTryCount) {
SeaTunnelConfig seaTunnelConfig =
ConfigProvider.locateAndGetSeaTunnelConfig();
@@ -320,6 +503,82 @@ public class CoordinatorServiceTest {
return SeaTunnelServerStarter.createHazelcastInstance(seaTunnelConfig);
}
+ private CoordinatorService newMockCoordinatorService(SeaTunnelServer
server) {
+ return newMockCoordinatorService(server, new EngineConfig());
+ }
+
+ private CoordinatorService newMockCoordinatorService(
+ SeaTunnelServer server, EngineConfig engineConfig) {
+ NodeEngineImpl nodeEngine = Mockito.mock(NodeEngineImpl.class);
+ ILogger logger = Mockito.mock(ILogger.class);
+ HazelcastInstanceImpl hazelcastInstance =
Mockito.mock(HazelcastInstanceImpl.class);
+ IMap<Object, Object> map = Mockito.mock(IMap.class);
+ Mockito.when(map.entrySet()).thenReturn(Collections.emptySet());
+
Mockito.when(nodeEngine.getLogger(Mockito.any(Class.class))).thenReturn(logger);
+
Mockito.when(nodeEngine.getHazelcastInstance()).thenReturn(hazelcastInstance);
+
Mockito.when(hazelcastInstance.getMap(Mockito.anyString())).thenReturn(map);
+
+ CoordinatorService coordinatorService =
+ new CoordinatorService(nodeEngine, server, engineConfig);
+ stopCoordinatorSchedulers(coordinatorService);
+ return coordinatorService;
+ }
+
+ private void stopCoordinatorSchedulers(CoordinatorService
coordinatorService) {
+ ReflectionUtils.getField(coordinatorService, "masterActiveListener")
+ .map(ScheduledExecutorService.class::cast)
+ .ifPresent(ScheduledExecutorService::shutdownNow);
+ ReflectionUtils.getField(coordinatorService,
"pipelineCleanupScheduler")
+ .map(ScheduledExecutorService.class::cast)
+ .ifPresent(ScheduledExecutorService::shutdownNow);
+ }
+
+ private JobMaster enqueueMockPendingJob(
+ CoordinatorService coordinatorService, long jobId, CountDownLatch
runLatch) {
+ return enqueueMockPendingJob(coordinatorService, jobId, runLatch,
true);
+ }
+
+ private JobMaster enqueueMockPendingJob(
+ CoordinatorService coordinatorService,
+ long jobId,
+ CountDownLatch runLatch,
+ boolean preApplyResourceResult) {
+ JobMaster jobMaster = Mockito.mock(JobMaster.class);
+ PhysicalPlan physicalPlan = Mockito.mock(PhysicalPlan.class);
+ @SuppressWarnings("unchecked")
+ PassiveCompletableFuture<JobResult> completionFuture =
+ Mockito.mock(PassiveCompletableFuture.class);
+
+ Mockito.when(jobMaster.getJobId()).thenReturn(jobId);
+
Mockito.when(jobMaster.preApplyResources()).thenReturn(preApplyResourceResult);
+ Mockito.when(jobMaster.getPhysicalPlan()).thenReturn(physicalPlan);
+
Mockito.when(jobMaster.getJobMasterCompleteFuture()).thenReturn(completionFuture);
+ Mockito.when(completionFuture.isCancelled()).thenReturn(false);
+
Mockito.when(completionFuture.isCompletedExceptionally()).thenReturn(false);
+ Mockito.when(physicalPlan.getJobFullName()).thenReturn("mock-job-" +
jobId);
+ Mockito.doAnswer(
+ invocation -> {
+ runLatch.countDown();
+ return null;
+ })
+ .when(jobMaster)
+ .run();
+
+ coordinatorService
+ .getPendingJobQueue()
+ .put(new PendingJobInfo(PendingSourceState.SUBMIT, jobMaster));
+ return jobMaster;
+ }
+
+ private ThreadPoolExecutor getCoordinatorExecutor(CoordinatorService
coordinatorService) {
+ return ReflectionUtils.getField(coordinatorService, "executorService")
+ .map(ThreadPoolExecutor.class::cast)
+ .orElseThrow(
+ () ->
+ new AssertionError(
+ "Failed to get coordinator
executorService by reflection"));
+ }
+
@Test
public void
testSeaTunnelEngineRetryableExceptionOperationCanBeRetryByHazelcast() {
@@ -795,6 +1054,13 @@ public class CoordinatorServiceTest {
method.invoke(coordinatorService, jobId, jobInfo);
}
+ private void invokeCheckNewActiveMaster(CoordinatorService
coordinatorService)
+ throws Exception {
+ Method method =
CoordinatorService.class.getDeclaredMethod("checkNewActiveMaster");
+ method.setAccessible(true);
+ method.invoke(coordinatorService);
+ }
+
@Test
public void testClearCoordinatorService() {
JobInformation jobInformation =
@@ -903,13 +1169,17 @@ public class CoordinatorServiceTest {
coordinatorService, "runningJobInfoIMap",
runningJobInfoIMap);
}
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+
coordinatorService.getPendingJobQueue().contains(jobId)
+ ||
getRunningJobMasterMap(coordinatorService)
+
.containsKey(jobId)));
Long[] jobStateTimestamps =
runningJobStateTimestampsIMap.get(jobId);
Assertions.assertNotNull(jobStateTimestamps);
Assertions.assertEquals(
initializationTimestamp,
jobStateTimestamps[JobStatus.INITIALIZING.ordinal()]);
- Assertions.assertTrue(
- coordinatorService.getPendingJobQueue().contains(jobId)
- ||
getRunningJobMasterMap(coordinatorService).containsKey(jobId));
} finally {
instance.shutdown();
}
