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-12032-5d7c81cdcb34867b4e963bc548d6d6eb16ed26a0 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 30170807f8d9d4608f03b294295129d17bb134f2 Author: David Zollo <[email protected]> AuthorDate: Thu Sep 3 05:12:48 2026 +0000 [Test][E2E] Add regression coverage for FAILED-pipeline metrics cleanup (#12032) --- .../seatunnel/engine/e2e/JobExecutionIT.java | 105 +++++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java index c20aeb8522..54d8d122c6 100644 --- a/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java +++ b/seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobExecutionIT.java @@ -27,7 +27,12 @@ import org.apache.seatunnel.engine.common.config.JobConfig; import org.apache.seatunnel.engine.common.config.SeaTunnelConfig; import org.apache.seatunnel.engine.common.job.JobResult; import org.apache.seatunnel.engine.common.job.JobStatus; +import org.apache.seatunnel.engine.server.SeaTunnelServer; import org.apache.seatunnel.engine.server.SeaTunnelServerStarter; +import org.apache.seatunnel.engine.server.dag.physical.PipelineLocation; +import org.apache.seatunnel.engine.server.execution.TaskGroupLocation; +import org.apache.seatunnel.engine.server.execution.TaskLocation; +import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; @@ -38,6 +43,8 @@ import com.hazelcast.client.config.ClientConfig; import com.hazelcast.instance.impl.HazelcastInstanceImpl; import lombok.extern.slf4j.Slf4j; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -160,6 +167,104 @@ public class JobExecutionIT { } } + /** + * Regression test for the FAILED-pipeline metrics leak fixed by <a + * href="https://github.com/apache/seatunnel/pull/10757";>#10757</a> ("[Fix][Zeta] Clean failed + * pipeline metrics without full-map cleanup scan"). Before that fix, {@code + * CoordinatorService#shouldCleanup} and {@code JobMaster#removeMetricsContext} only matched + * {@code PipelineStatus.CANCELED}/{@code FINISHED}, so a pipeline that ended in {@code FAILED} + * never had its entry removed from the coordinator's metrics {@code IMap}: it leaked there for + * the remaining life of the cluster, once per failed pipeline. + * + * <p>This reuses the same deterministic {@code NumberFormatException} failure as {@link + * #testGetErrorInfo()} (a SQL transform casting a random string to {@code int}), which fails + * the pipeline almost immediately after task deployment. That speed makes it unsafe to rely on + * the worker's real metrics-report cycle ({@code job-metrics-backup-interval}, 10 seconds by + * default) to populate the coordinator's metrics {@code IMap} before the pipeline ends: the + * pipeline can fail and have its task group context torn down before the next scheduled report + * ever fires, which would make "no metrics after FAILED" trivially true and prove nothing about + * the fix. Instead, this test seeds one metrics entry for the pipeline directly through {@link + * SeaTunnelServer#updateMetrics}, the exact entry point the worker's periodic reporter itself + * calls (via {@code ReportMetricsOperation}). From the coordinator cleanup logic's point of + * view, a seeded entry is indistinguishable from a worker-reported one: {@code + * CoordinatorService#shouldCleanup} and {@code #processPendingPipelineCleanup} match purely on + * {@link PipelineLocation} and the pipeline's terminal status recorded by the real job, not on + * how the metrics entry was populated. + * + * <p>Cleanup can complete either through the immediate best-effort call in {@code + * SubPlan#subPlanDone}, or, if that racing attempt lands before this test seeds the entry, + * through the 60-second scheduled {@code CoordinatorService#cleanupPendingPipelines} safety net + * introduced by <a href="https://github.com/apache/seatunnel/pull/10418";>#10418</a>. This test + * does not need to know which one fires; it only asserts the end-to-end guarantee that the + * entry does not survive forever. + */ + @Test + public void testFailedPipelineMetricsAreCleanedUp() + throws ExecutionException, InterruptedException { + Common.setDeployMode(DeployMode.CLIENT); + String filePath = TestUtils.getResource("batch_fakesource_to_console_error.conf"); + JobConfig jobConfig = new JobConfig(); + jobConfig.setName("fake_to_console_error_cleanup"); + ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig(); + clientConfig.setClusterName(TestUtils.getClusterName("JobExecutionIT")); + try (SeaTunnelClient engineClient = new SeaTunnelClient(clientConfig)) { + ClientJobExecutionEnvironment jobExecutionEnv = + engineClient.createExecutionContext(filePath, jobConfig, SEATUNNEL_CONFIG); + final ClientJobProxy clientJobProxy = jobExecutionEnv.execute(); + + // A config with no shuffle/split boundary always compiles down to exactly one + // pipeline, whose id is always 1 (see ExecutionPlanGenerator#normalizePipeline), so + // the location can be constructed directly instead of racing the failing pipeline to + // read it off the JobMaster before its bookkeeping is torn down. + PipelineLocation pipelineLocation = new PipelineLocation(clientJobProxy.getJobId(), 1); + + SeaTunnelServer seaTunnelServer = + hazelcastInstance.node.getNodeEngine().getService(SeaTunnelServer.SERVICE_NAME); + + // Sentinel task group id 999: this entry stands in for whatever a worker would have + // reported for this pipeline. It is deliberately not tied to any of the job's real + // task groups so seeding it cannot race, or depend on, the job's own execution. + TaskGroupLocation seededTaskGroupLocation = + new TaskGroupLocation( + pipelineLocation.getJobId(), pipelineLocation.getPipelineId(), 999L); + TaskLocation seededTaskLocation = new TaskLocation(seededTaskGroupLocation, 0, 0); + Map<TaskLocation, SeaTunnelMetricsContext> seededMetrics = new HashMap<>(); + seededMetrics.put(seededTaskLocation, new SeaTunnelMetricsContext()); + seaTunnelServer.updateMetrics(seededMetrics); + Assertions.assertTrue( + seaTunnelServer + .getEngineContext() + .getStateStores() + .metricsSnapshotStore() + .containsPipeline(pipelineLocation), + "Sanity check: seeded metrics entry must be visible before asserting cleanup"); + + CompletableFuture<JobStatus> completableFuture = + CompletableFuture.supplyAsync(clientJobProxy::waitForJobComplete); + await().atMost(300000, TimeUnit.MILLISECONDS) + .untilAsserted(() -> Assertions.assertTrue(completableFuture.isDone())); + + JobResult result = clientJobProxy.getJobResultCache(); + Assertions.assertEquals(JobStatus.FAILED, result.getStatus()); + + // Bound generously above CoordinatorService#PIPELINE_CLEANUP_INTERVAL_SECONDS (60s): + // the scheduled safety net's first run can land anywhere in that window relative to + // when this pipeline actually failed. + await().atMost(120, TimeUnit.SECONDS) + .untilAsserted( + () -> + Assertions.assertFalse( + seaTunnelServer + .getEngineContext() + .getStateStores() + .metricsSnapshotStore() + .containsPipeline(pipelineLocation), + "Metrics for a FAILED pipeline must eventually be " + + "removed from the coordinator's metrics " + + "IMap, not retained forever")); + } + } + @Test public void testValidJobNameInJobConfig() throws ExecutionException, InterruptedException { Common.setDeployMode(DeployMode.CLIENT);
