This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 df94fb07db [Bug][Zeta] Surface periodic checkpoint trigger failures
(#12448)
df94fb07db is described below
commit df94fb07dbd41e6f146793011d5ea42880fce71a
Author: hutiefang76 <[email protected]>
AuthorDate: Thu Oct 22 09:11:25 2026 -0700
[Bug][Zeta] Surface periodic checkpoint trigger failures (#12448)
---
.../server/checkpoint/CheckpointCoordinator.java | 15 +++-
.../CheckpointCoordinatorScheduledTriggerTest.java | 91 ++++++++++++++++++++++
2 files changed, 105 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
index d4dfa45436..a5267c2f1d 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java
@@ -548,7 +548,20 @@ public class CheckpointCoordinator {
protected void scheduleTriggerPendingCheckpoint(
CheckpointType checkpointType, long delayMills) {
scheduler.schedule(
- () -> tryTriggerPendingCheckpoint(checkpointType),
+ () -> {
+ try {
+ tryTriggerPendingCheckpoint(checkpointType);
+ } catch (Throwable e) {
+ // A thrown trigger would otherwise drop the scheduled
checkpoint chain
+ // silently; fail the coordinator so the job can
handle the error.
+ handleCoordinatorError(
+ String.format(
+ "Failed to trigger %s for job id: %s,
pipeline id: %s",
+ checkpointType, jobId, pipelineId),
+ e,
+ CheckpointCloseReason.CHECKPOINT_INSIDE_ERROR);
+ }
+ },
delayMills,
TimeUnit.MILLISECONDS);
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorScheduledTriggerTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorScheduledTriggerTest.java
new file mode 100644
index 0000000000..27e3ac1e46
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinatorScheduledTriggerTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.server.checkpoint;
+
+import org.apache.seatunnel.engine.checkpoint.storage.api.CheckpointStorage;
+import org.apache.seatunnel.engine.common.config.server.CheckpointConfig;
+import
org.apache.seatunnel.engine.common.config.server.CheckpointStorageConfig;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointIDCounter;
+import org.apache.seatunnel.engine.core.checkpoint.CheckpointType;
+import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import com.hazelcast.map.IMap;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+class CheckpointCoordinatorScheduledTriggerTest {
+
+ @Test
+ void unexpectedPeriodicTriggerErrorFailsCoordinator() throws Exception {
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ CheckpointManager checkpointManager =
Mockito.mock(CheckpointManager.class);
+ CheckpointConfig checkpointConfig = new CheckpointConfig();
+ checkpointConfig.setStorage(new CheckpointStorageConfig());
+ TaskLocation taskLocation = new TaskLocation(new TaskGroupLocation(1L,
1, 1), 1, 1);
+ CheckpointPlan plan =
+ CheckpointPlan.builder()
+ .pipelineId(1)
+ .pipelineSubtasks(Collections.singleton(taskLocation))
+ .startingSubtasks(Collections.singleton(taskLocation))
+ .build();
+ @SuppressWarnings("unchecked")
+ IMap<Object, Object> runningJobStateIMap = Mockito.mock(IMap.class);
+ Mockito.when(runningJobStateIMap.get(Mockito.any()))
+ .thenReturn(CheckpointCoordinatorStatus.RUNNING);
+ CheckpointCoordinator coordinator =
+ Mockito.spy(
+ new CheckpointCoordinator(
+ checkpointManager,
+ Mockito.mock(CheckpointStorage.class),
+ checkpointConfig,
+ 1L,
+ plan,
+ Mockito.mock(CheckpointIDCounter.class),
+ null,
+ executorService,
+ runningJobStateIMap,
+ false,
+ null));
+ try {
+ Mockito.doThrow(new IllegalStateException("periodic trigger
failed"))
+ .when(coordinator)
+
.tryTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE);
+
+
coordinator.scheduleTriggerPendingCheckpoint(CheckpointType.CHECKPOINT_TYPE, 0);
+
+ CheckpointCoordinatorState state =
+ coordinator.waitCheckpointCoordinatorComplete().get(5,
TimeUnit.SECONDS);
+ Assertions.assertEquals(
+ CheckpointCoordinatorStatus.FAILED,
state.getCheckpointCoordinatorStatus());
+ Assertions.assertTrue(state.getThrowableMsg().contains("periodic
trigger failed"));
+ Mockito.verify(checkpointManager, Mockito.timeout(5000))
+ .handleCheckpointError(1, false);
+ } finally {
+ coordinator.cancelCheckpoint();
+ executorService.shutdownNow();
+ }
+ }
+}
