This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11495-9e33353713a8b216130abe89481bcf05ed7c0ae7 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 886eb632fd278683bf4dc989d848ee4d14252b82 Author: Jast <[email protected]> AuthorDate: Sat Oct 24 15:46:26 2026 +0000 [Fix][Zeta] Reset metrics after pipeline recovery (#11495) Co-authored-by: zhangshenghang <[email protected]> Co-authored-by: zhangshenghang <[email protected]> Co-authored-by: jast <[email protected]> Co-authored-by: Claude Code <[email protected]> Co-authored-by: multica-agent <[email protected]> --- docs/en/engines/zeta/rest-api-job-lifecycle.md | 6 + .../introduction/concepts/incompatible-changes.md | 7 + docs/zh/engines/zeta/rest-api-job-lifecycle.md | 4 + .../introduction/concepts/incompatible-changes.md | 7 + .../engine/server/dag/physical/SubPlan.java | 1 + .../seatunnel/engine/server/master/JobMaster.java | 53 +++++- .../dag/physical/StateTransitionCleanupTest.java | 194 +++++++++++++++++++++ 7 files changed, 263 insertions(+), 9 deletions(-) diff --git a/docs/en/engines/zeta/rest-api-job-lifecycle.md b/docs/en/engines/zeta/rest-api-job-lifecycle.md index f0dcb302d6..16cca27875 100644 --- a/docs/en/engines/zeta/rest-api-job-lifecycle.md +++ b/docs/en/engines/zeta/rest-api-job-lifecycle.md @@ -185,6 +185,12 @@ Key metric fields: | `SourceReceivedQPS` | Current read throughput (rows/sec) | | `SinkWriteQPS` | Current write throughput (rows/sec) | +Counters are reported by the current attempt of each pipeline. When a pipeline is restored after a +failover (or retried after a failure), the new attempt starts counting from zero instead of +accumulating the values left behind by the failed attempt, for both batch and streaming jobs. A +worker report that was in flight during the restore can still carry a stale value for at most a +millisecond-scale window. + --- ## 4. Querying Job Logs diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 0decce7cd3..f8e3f2f95c 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -5,6 +5,13 @@ You need to check this document before you upgrade to related version. ## dev +### Zeta: runtime metric counters restart from zero after a pipeline failover restore + +- **Behavior change: runtime counters no longer accumulate values from a failed pipeline attempt** + - **Description**: When the Zeta engine restores a pipeline after a worker failover (or retries it after a failure), the new attempt's runtime counters (`SourceReceivedCount`, `SinkWriteCount`, queue metrics) now start from zero. Previously the replacement attempt continued from the failed attempt's values, because task locations are stable across retries and the new attempt's counters were seeded from the shared metrics store, so finished-job metrics could include work already done an [...] + - **Impact**: For batch jobs this removes over-counted finished-job metrics. For streaming jobs resumed after a failover, reported totals now restart at the new attempt's values instead of continuing the pre-failure totals; external dashboards or alerts that aggregate these counters must not read the reset as data loss. The reset is best effort: a worker report already in flight during the restore can still carry a stale value for at most a millisecond-scale window. + - **Migration Guide**: No action is required for correctness. If you need lifetime totals that survive failovers, accumulate them outside the running job, for example from the stored metrics of finished jobs in the job history. + ### JSON format: `ARRAY` and `MAP` columns reject values of the wrong container type When a field is declared as `ARRAY`, a JSON value that is not an array (a string, a number, a boolean or an object) previously produced an empty array `[]`, or an array holding `null` for every member of the object, without any warning. When a field is declared as `MAP`, a JSON value that is not an object silently produced an empty map `{}`. Both happened with `ignore-parse-errors` disabled as well as enabled. diff --git a/docs/zh/engines/zeta/rest-api-job-lifecycle.md b/docs/zh/engines/zeta/rest-api-job-lifecycle.md index f2d06f4180..ca8f4e45aa 100644 --- a/docs/zh/engines/zeta/rest-api-job-lifecycle.md +++ b/docs/zh/engines/zeta/rest-api-job-lifecycle.md @@ -184,6 +184,10 @@ curl http://<master>:8080/job-info/<jobId> | `SourceReceivedQPS` | 当前读取吞吐量（行/秒）| | `SinkWriteQPS` | 当前写入吞吐量（行/秒）| +计数器由每个管道当前一次执行上报。故障转移后管道被恢复（或失败后重试）时，新一次执行会从零开始计数， +不再累计失败执行残留的数值，批作业和流作业均如此。恢复过程中恰好在途的 worker 上报，最多可能在毫秒级 +窗口内携带一次旧数值。 + --- ## 4. 查询作业日志 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index aca33c6a98..9b9a4c3ec8 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -4,6 +4,13 @@ ## dev +### Zeta：管道故障转移恢复后运行时计数器从零重新计数 + +- **行为变更：运行时计数器不再累计失败管道执行的数值** + - **描述**：Zeta 引擎在 worker 故障转移后恢复管道（或失败后重试）时，新一次执行的运行时计数器（`SourceReceivedCount`、`SinkWriteCount`、队列指标）从零开始计数。此前由于任务位置在多次重试之间保持不变，新执行的计数器会从共享指标存储中继承失败执行的数值继续累加，导致已完成作业的指标可能包含早已被取消的执行所完成的工作量。 + - **影响**：对批作业，这消除了已完成作业指标多计的问题。对流作业，故障转移恢复后上报的总量会从新一次执行的数值重新开始，而不是延续故障前的累计值；外部聚合这些计数器的看板或告警不应将这种清零解读为数据丢失。该清零是尽力而为的：恢复期间恰好在途的 worker 上报，最多可能在毫秒级窗口内携带一次旧数值。 + - **迁移指南**：正确性上无需任何操作。如果需要跨故障转移的生命周期总量，请在运行中的作业之外自行累计，例如从作业历史中已完成作业存储的指标获取。 + ### JSON 格式：`ARRAY` 与 `MAP` 字段拒绝容器类型不匹配的值 当字段声明为 `ARRAY` 时，此前若 JSON 值不是数组（字符串、数字、布尔值或对象），会静默得到空数组 `[]`，若是对象则得到每个成员都为 `null` 的数组；当字段声明为 `MAP` 时，非对象的 JSON 值会静默得到空 Map `{}`。无论 `ignore-parse-errors` 关闭还是开启，行为都是如此。 diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java index 255bcbb112..fa4c2e3af0 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java @@ -494,6 +494,7 @@ public class SubPlan { String.format( "Restore time %s, pipeline %s", pipelineRestoreNum + "", pipelineFullName)); + jobMaster.clearPipelineMetricsForRestore(getPipelineLocation()); reset(); jobMaster.getCheckpointManager().reportedPipelineRunning(pipelineId, false); jobMaster.getPhysicalPlan().addPipelineEndCallback(this); diff --git a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java index d9068ddf40..836e017fc0 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java +++ b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java @@ -36,7 +36,6 @@ import org.apache.seatunnel.api.table.catalog.TablePath; import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; import org.apache.seatunnel.common.utils.ExceptionUtils; import org.apache.seatunnel.common.utils.RetryUtils; -import org.apache.seatunnel.common.utils.SeaTunnelException; import org.apache.seatunnel.engine.checkpoint.storage.PipelineState; import org.apache.seatunnel.engine.checkpoint.storage.api.CheckpointStorage; import org.apache.seatunnel.engine.common.Constant; @@ -1269,11 +1268,43 @@ public class JobMaster { && !checkpointManager.isPipelineSavePointEnd(pipelineLocation)) || pipelineStatus.equals(PipelineStatus.CANCELED)) { - try { - seaTunnelServer.removeMetrics(pipelineLocation); - } catch (Exception e) { - LOGGER.severe("failed to remove metrics", e); - } + removePipelineMetrics(pipelineLocation); + } + } + + /** + * Clears runtime metrics from a failed pipeline attempt before it is restored. + * + * <p>Task locations are stable across retries, so keeping the old snapshots would make the + * replacement attempt continue counting from the failed attempt's values. + * + * <p>The failed attempt's task-group contexts are cleaned first: a context that survives the + * wipe can re-populate the metrics store afterwards, defeating the reset. The reset is best + * effort: a worker report that was collected before its context was cleaned can still land + * right after the wipe, so at most such a millisecond-scale window can carry stale values. + * + * @param pipelineLocation pipeline that is about to be restored + */ + public void clearPipelineMetricsForRestore(PipelineLocation pipelineLocation) { + try { + cleanTaskGroupContext(pipelineLocation); + } catch (Exception e) { + LOGGER.warning( + "Failed to clean task-group contexts before restoring " + pipelineLocation, e); + } + removePipelineMetrics(pipelineLocation); + } + + /** + * Removes the pipeline's metrics from the metrics store. Best effort: a failure is logged and + * never propagated, so neither the terminal cleanup nor a pipeline restore is blocked. + */ + private void removePipelineMetrics(PipelineLocation pipelineLocation) { + try { + seaTunnelServer.removeMetrics(pipelineLocation); + } catch (Exception e) { + LOGGER.severe( + String.format("failed to remove metrics for pipeline %s", pipelineLocation), e); } } @@ -1294,13 +1325,17 @@ public class JobMaster { slotProfile.getWorker()) .get(); } - } catch (HazelcastInstanceNotActiveException e) { + } catch (Exception e) { + // Best effort per task group: one unreachable worker must not stop the + // remaining contexts from being cleaned, or those contexts would + // re-publish the failed attempt's metrics after the store was cleared. + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } LOGGER.warning( String.format( "%s clean TaskGroupContext with exception: %s.", taskGroupLocation, ExceptionUtils.getMessage(e))); - } catch (Exception e) { - throw new SeaTunnelException(e.getMessage()); } }); } diff --git a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java index 2bcf9a53db..b0bbd7204a 100644 --- a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java +++ b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/physical/StateTransitionCleanupTest.java @@ -26,26 +26,48 @@ import org.apache.seatunnel.engine.common.job.JobStatus; import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture; import org.apache.seatunnel.engine.core.dag.logical.LogicalDag; 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.AbstractSeaTunnelServerTest; +import org.apache.seatunnel.engine.server.SeaTunnelServer; import org.apache.seatunnel.engine.server.TestUtils; +import org.apache.seatunnel.engine.server.checkpoint.CheckpointManager; +import org.apache.seatunnel.engine.server.common.statestore.metrics.MetricsSnapshotStateStore; import org.apache.seatunnel.engine.server.execution.ExecutionState; import org.apache.seatunnel.engine.server.execution.TaskExecutionState; +import org.apache.seatunnel.engine.server.execution.TaskGroupLocation; +import org.apache.seatunnel.engine.server.execution.TaskLocation; +import org.apache.seatunnel.engine.server.master.JobHistoryService; +import org.apache.seatunnel.engine.server.master.JobMaster; +import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext; +import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager; +import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junitpioneer.jupiter.SetEnvironmentVariable; +import org.mockito.InOrder; +import com.hazelcast.cluster.Address; import com.hazelcast.map.IMap; +import java.lang.reflect.Method; import java.net.MalformedURLException; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import static org.apache.seatunnel.engine.common.config.server.QueueType.BLOCKINGQUEUE; import static org.apache.seatunnel.engine.core.classloader.DefaultClassLoaderService.SKIP_CHECK_JAR; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; @SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true") class StateTransitionCleanupTest extends AbstractSeaTunnelServerTest { @@ -157,6 +179,178 @@ class StateTransitionCleanupTest extends AbstractSeaTunnelServerTest { Assertions.assertEquals(JobStatus.FAILED, planWithStateMaps.runningJobState.get(jobId)); } + @Test + void testPipelineRestoreClearsMetricsFromTheFailedAttempt() throws Exception { + long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId(); + JobImmutableInformation jobImmutableInformation = mock(JobImmutableInformation.class); + JobConfig jobConfig = mock(JobConfig.class); + Map<String, Object> envOptions = new HashMap<>(); + envOptions.put("job.retry.times", "1"); + envOptions.put("job.retry.interval.seconds", "0"); + when(jobImmutableInformation.getJobId()).thenReturn(jobId); + when(jobImmutableInformation.getJobConfig()).thenReturn(jobConfig); + when(jobConfig.getEnvOptions()).thenReturn(envOptions); + + IMap<Object, Object> runningJobState = + nodeEngine.getHazelcastInstance().getMap("restoreMetricsState-" + jobId); + IMap<Object, Long[]> runningJobStateTimestamps = + nodeEngine.getHazelcastInstance().getMap("restoreMetricsStateTimestamps-" + jobId); + PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1); + runningJobState.put(pipelineLocation, PipelineStatus.FAILED); + + ExecutorService executorService = Executors.newSingleThreadExecutor(); + try { + SubPlan subPlan = + new SubPlan( + 1, + 1, + System.currentTimeMillis(), + Collections.emptyList(), + Collections.emptyList(), + jobImmutableInformation, + executorService, + runningJobState, + runningJobStateTimestamps, + Collections.emptyMap()); + JobMaster jobMaster = mock(JobMaster.class); + CheckpointManager checkpointManager = mock(CheckpointManager.class); + PhysicalPlan physicalPlan = mock(PhysicalPlan.class); + when(jobMaster.getCheckpointManager()).thenReturn(checkpointManager); + when(jobMaster.getPhysicalPlan()).thenReturn(physicalPlan); + subPlan.setJobMaster(jobMaster); + + Method restore = SubPlan.class.getDeclaredMethod("prepareRestorePipeline"); + restore.setAccessible(true); + restore.invoke(subPlan); + + verify(jobMaster).clearPipelineMetricsForRestore(pipelineLocation); + } finally { + executorService.shutdownNow(); + } + } + + /** + * The restore clear must clean the failed attempt's task-group contexts before it removes the + * pipeline's metrics: a context that survives the wipe re-publishes the old counters during the + * next worker report and defeats the reset. + */ + @Test + void testRestoreClearRunsTaskGroupCleanupBeforeMetricsRemoval() { + long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId(); + PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1); + Map<TaskGroupLocation, SlotProfile> slotProfiles = newOwnedSlotProfiles(jobId); + @SuppressWarnings("unchecked") + IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>> ownedSlotProfilesIMap = + mock(IMap.class); + when(ownedSlotProfilesIMap.get(pipelineLocation)).thenReturn(slotProfiles); + SeaTunnelServer seaTunnelServer = mock(SeaTunnelServer.class); + + JobMaster jobMaster = newJobMaster(jobId, ownedSlotProfilesIMap, seaTunnelServer); + jobMaster.clearPipelineMetricsForRestore(pipelineLocation); + + InOrder inOrder = inOrder(ownedSlotProfilesIMap, seaTunnelServer); + inOrder.verify(ownedSlotProfilesIMap).get(pipelineLocation); + inOrder.verify(seaTunnelServer).removeMetrics(pipelineLocation); + } + + /** The restore clear must remove the failed attempt's snapshots from the real state store. */ + @Test + void testRestoreClearRemovesMetricsFromTheStateStore() { + long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId(); + PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1); + PipelineLocation otherPipelineLocation = new PipelineLocation(jobId, 2); + upsertMetricsForPipeline(pipelineLocation); + upsertMetricsForPipeline(otherPipelineLocation); + Assertions.assertTrue(hasMetricsForPipeline(pipelineLocation)); + Assertions.assertTrue(hasMetricsForPipeline(otherPipelineLocation)); + + IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>> ownedSlotProfilesIMap = + nodeEngine.getHazelcastInstance().getMap("restoreOwnedSlotProfiles-" + jobId); + ownedSlotProfilesIMap.put(pipelineLocation, newOwnedSlotProfiles(jobId)); + + JobMaster jobMaster = newJobMaster(jobId, ownedSlotProfilesIMap, server); + jobMaster.clearPipelineMetricsForRestore(pipelineLocation); + + Assertions.assertFalse( + hasMetricsForPipeline(pipelineLocation), + "Restoring a pipeline must not keep the failed attempt's metrics"); + Assertions.assertTrue( + hasMetricsForPipeline(otherPipelineLocation), + "The clear must not touch other pipelines of the same job"); + } + + @Test + void testMetricsCleanupFailureDoesNotPreventRestore() { + long jobId = instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId(); + PipelineLocation location = new PipelineLocation(jobId, 1); + SeaTunnelServer failingServer = mock(SeaTunnelServer.class); + doThrow(new IllegalStateException("metrics map unavailable")) + .when(failingServer) + .removeMetrics(location); + IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>> emptyOwnedSlotProfiles = + nodeEngine.getHazelcastInstance().getMap("restoreOwnedSlotProfilesEmpty-" + jobId); + JobMaster jobMaster = newJobMaster(jobId, emptyOwnedSlotProfiles, failingServer); + + Assertions.assertDoesNotThrow(() -> jobMaster.clearPipelineMetricsForRestore(location)); + verify(failingServer).removeMetrics(location); + } + + private Map<TaskGroupLocation, SlotProfile> newOwnedSlotProfiles(long jobId) { + Address localAddress = instance.getCluster().getLocalMember().getAddress(); + Map<TaskGroupLocation, SlotProfile> slotProfiles = new HashMap<>(); + slotProfiles.put( + new TaskGroupLocation(jobId, 1, 1L), + new SlotProfile(localAddress, 0, null, "restore-clear-test")); + return slotProfiles; + } + + /** + * Creates a JobMaster with only the pieces {@code clearPipelineMetricsForRestore} touches: a + * real node engine backed by the test cluster and the given owned-slot-profiles map and server. + * The remaining constructor arguments are never used by this code path. + */ + private JobMaster newJobMaster( + long jobId, + IMap<PipelineLocation, Map<TaskGroupLocation, SlotProfile>> ownedSlotProfilesIMap, + SeaTunnelServer seaTunnelServer) { + IMap<Object, Object> runningJobState = + nodeEngine.getHazelcastInstance().getMap("restoreJobMasterState-" + jobId); + IMap<Object, Object> runningJobStateTimestamps = + nodeEngine.getHazelcastInstance().getMap("restoreJobMasterStateTs-" + jobId); + IMap<Long, JobInfo> runningJobInfo = + nodeEngine.getHazelcastInstance().getMap("restoreJobMasterInfo-" + jobId); + return new JobMaster( + jobId, + nodeEngine.getSerializationService().toData("unused"), + nodeEngine, + mock(ExecutorService.class), + mock(ResourceManager.class), + mock(JobHistoryService.class), + runningJobState, + runningJobStateTimestamps, + ownedSlotProfilesIMap, + runningJobInfo, + new EngineConfig(), + seaTunnelServer); + } + + private void upsertMetricsForPipeline(PipelineLocation pipelineLocation) { + TaskGroupLocation taskGroupLocation = + new TaskGroupLocation( + pipelineLocation.getJobId(), pipelineLocation.getPipelineId(), 1L); + TaskLocation taskLocation = new TaskLocation(taskGroupLocation, 0, 0); + + Map<TaskLocation, SeaTunnelMetricsContext> local = new HashMap<>(); + local.put(taskLocation, new SeaTunnelMetricsContext()); + server.updateMetrics(local); + } + + private boolean hasMetricsForPipeline(PipelineLocation pipelineLocation) { + MetricsSnapshotStateStore metricsStore = + server.getEngineContext().getStateStores().metricsSnapshotStore(); + return metricsStore.containsPipeline(pipelineLocation); + } + private PlanWithStateMaps createPhysicalPlan(long jobId) throws MalformedURLException { JobContext jobContext = new JobContext(jobId); jobContext.setJobMode(JobMode.BATCH);
