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 6ee8f8d936 [Fix][Zeta] Clean failed pipeline metrics without full-map
cleanup scan (#10757)
6ee8f8d936 is described below
commit 6ee8f8d9368ac77eb3d22961ee914f0d602a6c7a
Author: David Zollo <[email protected]>
AuthorDate: Tue Jul 21 23:11:27 2026 +0800
[Fix][Zeta] Clean failed pipeline metrics without full-map cleanup scan
(#10757)
Co-authored-by: Doyeon Kim <[email protected]>
Co-authored-by: Daniel <[email protected]>
Co-authored-by: Daniel <[email protected]>
Co-authored-by: DanielLeens <[email protected]>
---
.../engine/common/config/EngineConfig.java | 2 +-
.../engine/server/CoordinatorService.java | 177 +++++++++++++------
.../HazelcastMetricsSnapshotStateStore.java | 52 ++----
.../seatunnel/engine/server/master/JobMaster.java | 8 +-
.../master/cleanup/PipelineCleanupRecord.java | 12 +-
.../CoordinatorServicePipelineCleanupTest.java | 193 +++++++++++++++++++--
.../engine/server/master/JobMasterTest.java | 61 ++++++-
.../master/cleanup/PipelineCleanupRecordTest.java | 9 +-
8 files changed, 406 insertions(+), 108 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java
index dec7eb8190..82b7592575 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java
@@ -169,7 +169,7 @@ public class EngineConfig {
public void setJobMetricsPartitionCount(int jobMetricsPartitionCount) {
checkPositive(
jobMetricsPartitionCount,
-
ServerConfigOptions.MasterServerConfigOptions.JOB_METRICS_PARTITION_COUNT
+
ServerConfigOptions.MasterServerConfigOptions.JOB_METRICS_PARTITION_COUNT.key()
+ " must be > 0");
this.jobMetricsPartitionCount = jobMetricsPartitionCount;
}
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 df2134cd98..0d33d82757 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
@@ -602,71 +602,94 @@ public class CoordinatorService {
}
}
- private void processPendingPipelineCleanup(
+ @VisibleForTesting
+ void processPendingPipelineCleanup(
PipelineLocation pipelineLocation, PipelineCleanupRecord record) {
if (pipelineLocation == null || record == null) {
return;
}
- if (!shouldCleanup(record)) {
- removePendingCleanupRecord(pipelineLocation, record);
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+ pendingPipelineCleanupIMap;
+ if (pendingCleanupIMap == null) {
return;
}
- PipelineStatus currentStatus =
getPipelineStatusFromIMap(pipelineLocation);
- if (currentStatus != null && !currentStatus.isEndState()) {
- return;
- }
+ boolean locked = false;
+ try {
+ pendingCleanupIMap.lock(pipelineLocation);
+ locked = true;
- long now = System.currentTimeMillis();
- PipelineCleanupRecord updated = copy(record);
- updated.setLastAttemptTimeMillis(now);
- updated.setAttemptCount(record.getAttemptCount() + 1);
+ // entrySet() is a snapshot. Revalidate it while holding the same
key lock used by
+ // restore invalidation so an old execution round cannot clean a
newer round.
+ PipelineCleanupRecord currentRecord =
pendingCleanupIMap.get(pipelineLocation);
+ if (!record.equals(currentRecord)) {
+ return;
+ }
+ if (!shouldCleanup(record)) {
+ removePendingCleanupRecord(pipelineLocation, record);
+ return;
+ }
- if (!updated.isMetricsImapCleaned() &&
cleanupPipelineMetrics(pipelineLocation)) {
- updated.setMetricsImapCleaned(true);
- }
+ PipelineStatus currentStatus =
getPipelineStatusFromIMap(pipelineLocation);
+ if (currentStatus != null && !currentStatus.isEndState()) {
+ return;
+ }
- Map<TaskGroupLocation, Address> taskGroups = updated.getTaskGroups();
- if (taskGroups != null && !taskGroups.isEmpty()) {
- for (Map.Entry<TaskGroupLocation, Address> taskGroup :
taskGroups.entrySet()) {
- TaskGroupLocation taskGroupLocation = taskGroup.getKey();
- if (updated.getCleanedTaskGroups() != null
- &&
updated.getCleanedTaskGroups().contains(taskGroupLocation)) {
- continue;
- }
- Address workerAddress = taskGroup.getValue();
- if (workerAddress == null
- ||
nodeEngine.getClusterService().getMember(workerAddress) == null) {
- continue;
- }
- try {
- NodeEngineUtil.sendOperationToMemberNode(
- nodeEngine,
- new
CleanTaskGroupContextOperation(taskGroupLocation),
- workerAddress)
- .get();
- updated.getCleanedTaskGroups().add(taskGroupLocation);
- } catch (HazelcastInstanceNotActiveException e) {
- logger.warning(
- String.format(
- "%s clean TaskGroupContext failed: %s",
- taskGroupLocation,
ExceptionUtils.getMessage(e)));
- } catch (Exception e) {
- logger.warning(
- String.format(
- "%s clean TaskGroupContext failed: %s",
- taskGroupLocation,
ExceptionUtils.getMessage(e)),
- e);
+ long now = System.currentTimeMillis();
+ PipelineCleanupRecord updated = copy(record);
+ updated.setLastAttemptTimeMillis(now);
+ updated.setAttemptCount(record.getAttemptCount() + 1);
+
+ if (!updated.isMetricsImapCleaned() &&
cleanupPipelineMetrics(pipelineLocation)) {
+ updated.setMetricsImapCleaned(true);
+ }
+
+ Map<TaskGroupLocation, Address> taskGroups =
updated.getTaskGroups();
+ if (taskGroups != null && !taskGroups.isEmpty()) {
+ for (Map.Entry<TaskGroupLocation, Address> taskGroup :
taskGroups.entrySet()) {
+ TaskGroupLocation taskGroupLocation = taskGroup.getKey();
+ if (updated.getCleanedTaskGroups() != null
+ &&
updated.getCleanedTaskGroups().contains(taskGroupLocation)) {
+ continue;
+ }
+ Address workerAddress = taskGroup.getValue();
+ if (workerAddress == null
+ ||
nodeEngine.getClusterService().getMember(workerAddress) == null) {
+ continue;
+ }
+ try {
+ NodeEngineUtil.sendOperationToMemberNode(
+ nodeEngine,
+ new
CleanTaskGroupContextOperation(taskGroupLocation),
+ workerAddress)
+ .get();
+ updated.getCleanedTaskGroups().add(taskGroupLocation);
+ } catch (HazelcastInstanceNotActiveException e) {
+ logger.warning(
+ String.format(
+ "%s clean TaskGroupContext failed: %s",
+ taskGroupLocation,
ExceptionUtils.getMessage(e)));
+ } catch (Exception e) {
+ logger.warning(
+ String.format(
+ "%s clean TaskGroupContext failed: %s",
+ taskGroupLocation,
ExceptionUtils.getMessage(e)),
+ e);
+ }
}
}
- }
- boolean replaced =
pendingPipelineCleanupIMap.replace(pipelineLocation, record, updated);
- if (!replaced) {
- return;
- }
- if (updated.isCleaned()) {
- pendingPipelineCleanupIMap.remove(pipelineLocation, updated);
+ boolean replaced = pendingCleanupIMap.replace(pipelineLocation,
record, updated);
+ if (!replaced) {
+ return;
+ }
+ if (updated.isCleaned()) {
+ pendingCleanupIMap.remove(pipelineLocation, updated);
+ }
+ } finally {
+ if (locked) {
+ pendingCleanupIMap.unlock(pipelineLocation);
+ }
}
}
@@ -690,7 +713,8 @@ public class CoordinatorService {
if (record.isSavepointEnd()) {
return false;
}
- return PipelineStatus.CANCELED.equals(record.getFinalStatus())
+ return PipelineStatus.FAILED.equals(record.getFinalStatus())
+ || PipelineStatus.CANCELED.equals(record.getFinalStatus())
|| PipelineStatus.FINISHED.equals(record.getFinalStatus());
}
@@ -1054,6 +1078,50 @@ public class CoordinatorService {
runningJobInfoIMap.remove(jobId);
}
+ /**
+ * Removes pending pipeline cleanup records from an older execution round
of the same job.
+ *
+ * <p>A savepoint-start submission reuses the same job id and pipeline
ids, while pipeline
+ * cleanup records are keyed only by {@link PipelineLocation}. If a
previous FAILED record
+ * survives into the restored round, it can delete metrics that the later
savepoint-end round
+ * must retain.
+ *
+ * @param jobId job id that is entering a new savepoint-start execution
round
+ */
+ @VisibleForTesting
+ void cleanupPendingPipelineCleanupForRestore(long jobId) {
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+ pendingPipelineCleanupIMap;
+ if (pendingCleanupIMap == null || pendingCleanupIMap.isEmpty()) {
+ return;
+ }
+ for (Map.Entry<PipelineLocation, PipelineCleanupRecord> entry :
+ pendingCleanupIMap.entrySet()) {
+ PipelineLocation pipelineLocation = entry.getKey();
+ if (pipelineLocation == null) {
+ continue;
+ }
+ boolean locked = false;
+ try {
+ pendingCleanupIMap.lock(pipelineLocation);
+ locked = true;
+ PipelineCleanupRecord currentRecord =
pendingCleanupIMap.get(pipelineLocation);
+ boolean sameJob =
+ pipelineLocation.getJobId() == jobId
+ || (currentRecord != null
+ && currentRecord.getPipelineLocation()
!= null
+ &&
currentRecord.getPipelineLocation().getJobId() == jobId);
+ if (sameJob) {
+ pendingCleanupIMap.remove(pipelineLocation);
+ }
+ } finally {
+ if (locked) {
+ pendingCleanupIMap.unlock(pipelineLocation);
+ }
+ }
+ }
+ }
+
/**
* Polls master ownership and reconciles the local coordinator lifecycle
with cluster state.
*
@@ -1201,6 +1269,9 @@ public class CoordinatorService {
JobMaster jobMaster = null;
JobInfo submittedJobInfo = null;
try {
+ if (isStartWithSavePoint) {
+ cleanupPendingPipelineCleanupForRestore(jobId);
+ }
JobCleanupRecord pendingCleanupRecord =
pendingJobCleanupIMap != null
? pendingJobCleanupIMap.get(jobId)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
index be4e3f83eb..20df3b848b 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java
@@ -30,14 +30,12 @@ import com.hazelcast.map.listener.EntryUpdatedListener;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
/** Implementation backed by a partitioned Hazelcast metrics {@link IMap}. */
public class HazelcastMetricsSnapshotStateStore
@@ -115,40 +113,24 @@ public class HazelcastMetricsSnapshotStateStore
@Override
public void removePipeline(final PipelineLocation pipelineLocation) {
- Map<Long, List<TaskLocation>> partitionedTasks = new HashMap<>();
- for (Map.Entry<Long, Map<TaskLocation, SeaTunnelMetricsContext>> entry
:
- metricsImap.entrySet()) {
- long partition = entry.getKey();
- List<TaskLocation> tasksToRemove =
- entry.getValue().keySet().stream()
- .filter(
- t ->
- t.getTaskGroupLocation()
- .getPipelineLocation()
- .equals(pipelineLocation))
- .collect(Collectors.toList());
- if (!tasksToRemove.isEmpty()) {
- partitionedTasks.put(partition, tasksToRemove);
- }
+ for (long partition = 0; partition < partitionCount; partition++) {
+ metricsImap.compute(
+ partition,
+ (ignored, current) -> {
+ if (current == null || current.isEmpty()) {
+ return current;
+ }
+ Map<TaskLocation, SeaTunnelMetricsContext> updated =
new HashMap<>(current);
+ updated.entrySet()
+ .removeIf(
+ entry ->
+ pipelineLocation.equals(
+ entry.getKey()
+
.getTaskGroupLocation()
+
.getPipelineLocation()));
+ return updated.isEmpty() ? null : updated;
+ });
}
-
- partitionedTasks
- .entrySet()
- .parallelStream()
- .forEach(
- entry -> {
- long partition = entry.getKey();
- List<TaskLocation> tasks = entry.getValue();
- metricsImap.compute(
- partition,
- (k, oldVal) -> {
- if (oldVal != null) {
- tasks.forEach(oldVal::remove);
- if (oldVal.isEmpty()) return null;
- }
- return oldVal;
- });
- });
}
@Override
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 9272f9a0bd..824c38256a 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
@@ -1114,8 +1114,11 @@ public class JobMaster {
PipelineStatus.FINISHED.equals(pipelineStatus)
&& checkpointManager != null
&&
checkpointManager.isPipelineSavePointEnd(pipelineLocation);
+ // Failed pipelines also need cleanup so their distributed metrics do
not leak into later
+ // task recovery or re-submission flows.
boolean shouldCleanup =
- PipelineStatus.CANCELED.equals(pipelineStatus)
+ PipelineStatus.FAILED.equals(pipelineStatus)
+ || PipelineStatus.CANCELED.equals(pipelineStatus)
|| (PipelineStatus.FINISHED.equals(pipelineStatus) &&
!savepointEnd);
if (!shouldCleanup) {
return;
@@ -1172,7 +1175,8 @@ public class JobMaster {
public void removeMetricsContext(
PipelineLocation pipelineLocation, PipelineStatus pipelineStatus) {
- if ((pipelineStatus.equals(PipelineStatus.FINISHED)
+ if (pipelineStatus.equals(PipelineStatus.FAILED)
+ || (pipelineStatus.equals(PipelineStatus.FINISHED)
&&
!checkpointManager.isPipelineSavePointEnd(pipelineLocation))
|| pipelineStatus.equals(PipelineStatus.CANCELED)) {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecord.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecord.java
index 6ace1cbb0b..b07b565581 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecord.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecord.java
@@ -136,6 +136,16 @@ public class PipelineCleanupRecord implements
IdentifiedDataSerializable {
&& cleanedTaskGroups.containsAll(taskGroups.keySet());
}
+ /**
+ * Merges a newer cleanup intent into this record while preserving
completed cleanup progress.
+ *
+ * <p>The caller passes the newer terminal-state record as {@code other}.
Its final status must
+ * win so an older FAILED or CANCELED intent cannot keep driving cleanup
after a later terminal
+ * state is reported for the same pipeline location.
+ *
+ * @param other newer cleanup record for the same pipeline location
+ * @return merged cleanup record
+ */
public PipelineCleanupRecord mergeFrom(PipelineCleanupRecord other) {
if (other == null) {
return this;
@@ -161,7 +171,7 @@ public class PipelineCleanupRecord implements
IdentifiedDataSerializable {
this.pipelineLocation != null
? this.pipelineLocation
: other.pipelineLocation,
- this.finalStatus != null ? this.finalStatus :
other.finalStatus,
+ other.finalStatus != null ? other.finalStatus :
this.finalStatus,
this.savepointEnd || other.savepointEnd,
mergedTaskGroups,
mergedCleaned,
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
index e8218e46ba..267596a7ce 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServicePipelineCleanupTest.java
@@ -36,7 +36,10 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.awaitility.Awaitility.await;
@@ -44,8 +47,7 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
@Test
void testCleanupRemovesMetricsAndRecordWhenNoTaskGroups() {
- CoordinatorService coordinatorService = server.getCoordinatorService();
- awaitCoordinatorActive(coordinatorService);
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
long jobId = System.currentTimeMillis();
PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
@@ -82,10 +84,48 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
Assertions.assertFalse(pendingCleanupIMap.containsKey(pipelineLocation));
}
+ @Test
+ void testCleanupRemovesMetricsAndRecordForFailedPipeline() {
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
+
+ long jobId = System.currentTimeMillis();
+ PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
+ PipelineLocation otherPipelineLocation = new PipelineLocation(jobId +
1, 1);
+
+ upsertMetricsForPipeline(pipelineLocation);
+ upsertMetricsForPipeline(otherPipelineLocation);
+ Assertions.assertTrue(hasMetricsForPipeline(pipelineLocation));
+ Assertions.assertTrue(hasMetricsForPipeline(otherPipelineLocation));
+
+ IMap<Object, Object> runningJobStateIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_STATE);
+ runningJobStateIMap.put(pipelineLocation, PipelineStatus.FAILED);
+
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_PENDING_PIPELINE_CLEANUP);
+ pendingCleanupIMap.put(
+ pipelineLocation,
+ new PipelineCleanupRecord(
+ pipelineLocation,
+ PipelineStatus.FAILED,
+ false,
+ Collections.emptyMap(),
+ Collections.emptySet(),
+ false,
+ System.currentTimeMillis(),
+ 0L,
+ 0));
+
+ coordinatorService.runPendingPipelineCleanupOnce();
+
+ Assertions.assertFalse(hasMetricsForPipeline(pipelineLocation));
+ Assertions.assertTrue(hasMetricsForPipeline(otherPipelineLocation));
+
Assertions.assertFalse(pendingCleanupIMap.containsKey(pipelineLocation));
+ }
+
@Test
void testSkipCleanupWhenPipelineNotEndState() {
- CoordinatorService coordinatorService = server.getCoordinatorService();
- awaitCoordinatorActive(coordinatorService);
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
long jobId = System.currentTimeMillis();
PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
@@ -122,8 +162,7 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
@Test
void testRemoveRecordWhenShouldCleanupIsFalse() {
- CoordinatorService coordinatorService = server.getCoordinatorService();
- awaitCoordinatorActive(coordinatorService);
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
long jobId = System.currentTimeMillis();
PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
@@ -157,10 +196,104 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
"Should not clean metrics when record is removed due to
shouldCleanup=false");
}
+ /**
+ * Verifies that savepoint-start invalidates stale failed cleanup records
from the previous
+ * round.
+ */
+ @Test
+ void testRestoreStartRemovesStaleFailedCleanupRecordForSameJob() {
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
+
+ long jobId = System.currentTimeMillis();
+ PipelineLocation restoredPipelineLocation = new
PipelineLocation(jobId, 1);
+ PipelineLocation otherPipelineLocation = new PipelineLocation(jobId +
1, 1);
+
+ upsertMetricsForPipeline(restoredPipelineLocation);
+ upsertMetricsForPipeline(otherPipelineLocation);
+ Assertions.assertTrue(hasMetricsForPipeline(restoredPipelineLocation));
+ Assertions.assertTrue(hasMetricsForPipeline(otherPipelineLocation));
+
+ IMap<Object, Object> runningJobStateIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_STATE);
+ runningJobStateIMap.put(restoredPipelineLocation,
PipelineStatus.FINISHED);
+ runningJobStateIMap.put(otherPipelineLocation, PipelineStatus.FAILED);
+
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_PENDING_PIPELINE_CLEANUP);
+ pendingCleanupIMap.put(
+ restoredPipelineLocation,
+ newPipelineCleanupRecord(restoredPipelineLocation,
PipelineStatus.FAILED, false));
+ pendingCleanupIMap.put(
+ otherPipelineLocation,
+ newPipelineCleanupRecord(otherPipelineLocation,
PipelineStatus.FAILED, false));
+
+ coordinatorService.cleanupPendingPipelineCleanupForRestore(jobId);
+
+
Assertions.assertFalse(pendingCleanupIMap.containsKey(restoredPipelineLocation));
+
Assertions.assertTrue(pendingCleanupIMap.containsKey(otherPipelineLocation));
+
+ coordinatorService.runPendingPipelineCleanupOnce();
+
+ Assertions.assertTrue(
+ hasMetricsForPipeline(restoredPipelineLocation),
+ "Stale failed cleanup record must not delete savepoint-end
metrics");
+ Assertions.assertFalse(hasMetricsForPipeline(otherPipelineLocation));
+
Assertions.assertFalse(pendingCleanupIMap.containsKey(otherPipelineLocation));
+ }
+
+ @Test
+ void
testRestoreInvalidationPreventsCapturedCleanupFromDeletingNewRoundMetrics()
+ throws Exception {
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
+
+ long jobId = System.currentTimeMillis();
+ PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
+ PipelineCleanupRecord staleRecord =
+ newPipelineCleanupRecord(pipelineLocation,
PipelineStatus.FAILED, false);
+
+ IMap<Object, Object> runningJobStateIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_RUNNING_JOB_STATE);
+ runningJobStateIMap.put(pipelineLocation, PipelineStatus.FAILED);
+
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_PENDING_PIPELINE_CLEANUP);
+ pendingCleanupIMap.put(pipelineLocation, staleRecord);
+ upsertMetricsForPipeline(pipelineLocation);
+
+ CountDownLatch cleanupStarted = new CountDownLatch(1);
+ pendingCleanupIMap.lock(pipelineLocation);
+ CompletableFuture<Void> capturedCleanup;
+ try {
+ capturedCleanup =
+ CompletableFuture.runAsync(
+ () -> {
+ cleanupStarted.countDown();
+
coordinatorService.processPendingPipelineCleanup(
+ pipelineLocation, staleRecord);
+ });
+ Assertions.assertTrue(cleanupStarted.await(10, TimeUnit.SECONDS));
+ Assertions.assertFalse(capturedCleanup.isDone());
+
+ coordinatorService.cleanupPendingPipelineCleanupForRestore(jobId);
+
Assertions.assertFalse(pendingCleanupIMap.containsKey(pipelineLocation));
+
+ // Simulate metrics published by the restored execution before the
captured cleanup
+ // gets CPU time again.
+ upsertMetricsForPipeline(pipelineLocation);
+ runningJobStateIMap.put(pipelineLocation, PipelineStatus.FINISHED);
+ } finally {
+ pendingCleanupIMap.unlock(pipelineLocation);
+ }
+
+ capturedCleanup.get(10, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ hasMetricsForPipeline(pipelineLocation),
+ "Captured cleanup from the old round must not delete
restored-round metrics");
+ }
+
@Test
void testCleanupUpdatesRecordAndKeepsItWhenTaskGroupCannotBeCleaned() {
- CoordinatorService coordinatorService = server.getCoordinatorService();
- awaitCoordinatorActive(coordinatorService);
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
long jobId = System.currentTimeMillis();
PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
@@ -203,8 +336,7 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
@Test
void testCleanupRemovesRecordWhenAllTaskGroupsCleaned() {
- CoordinatorService coordinatorService = server.getCoordinatorService();
- awaitCoordinatorActive(coordinatorService);
+ CoordinatorService coordinatorService =
awaitActiveCoordinatorService();
long jobId = System.currentTimeMillis();
PipelineLocation pipelineLocation = new PipelineLocation(jobId, 1);
@@ -252,15 +384,50 @@ class CoordinatorServicePipelineCleanupTest extends
AbstractSeaTunnelServerTest
server.updateMetrics(local);
}
+ /**
+ * Creates a focused pending cleanup record without task-group cleanup
requirements.
+ *
+ * <p>The cleanup tests use this helper to keep assertions focused on
metric cleanup behavior.
+ */
+ private PipelineCleanupRecord newPipelineCleanupRecord(
+ PipelineLocation pipelineLocation,
+ PipelineStatus pipelineStatus,
+ boolean savepointEnd) {
+ return new PipelineCleanupRecord(
+ pipelineLocation,
+ pipelineStatus,
+ savepointEnd,
+ Collections.emptyMap(),
+ Collections.emptySet(),
+ false,
+ System.currentTimeMillis(),
+ 0L,
+ 0);
+ }
+
private boolean hasMetricsForPipeline(PipelineLocation pipelineLocation) {
MetricsSnapshotStateStore metricsStore =
server.getEngineContext().getStateStores().metricsSnapshotStore();
return metricsStore.containsPipeline(pipelineLocation);
}
- private void awaitCoordinatorActive(CoordinatorService coordinatorService)
{
- await().atMost(30, TimeUnit.SECONDS)
+ /**
+ * Waits until the test server exposes an active coordinator service.
+ *
+ * <p>{@link SeaTunnelServer#getCoordinatorService()} throws before
active-master initialization
+ * finishes, so tests must wait around the lookup itself instead of
checking activity after the
+ * lookup succeeds.
+ */
+ private CoordinatorService awaitActiveCoordinatorService() {
+ AtomicReference<CoordinatorService> coordinatorServiceRef = new
AtomicReference<>();
+ await().ignoreExceptions()
+ .atMost(30, TimeUnit.SECONDS)
.untilAsserted(
- () ->
Assertions.assertTrue(coordinatorService.isCoordinatorActive()));
+ () -> {
+ CoordinatorService coordinatorService =
server.getCoordinatorService();
+
Assertions.assertTrue(coordinatorService.isCoordinatorActive());
+ coordinatorServiceRef.set(coordinatorService);
+ });
+ return coordinatorServiceRef.get();
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
index 8c66837036..a394b9980b 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/JobMasterTest.java
@@ -35,6 +35,9 @@ import
org.apache.seatunnel.engine.server.dag.physical.PipelineLocation;
import org.apache.seatunnel.engine.server.dag.physical.SubPlan;
import
org.apache.seatunnel.engine.server.dag.physical.UnknownPhysicalPlanException;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
+import org.apache.seatunnel.engine.server.execution.TaskLocation;
+import org.apache.seatunnel.engine.server.master.cleanup.PipelineCleanupRecord;
+import org.apache.seatunnel.engine.server.metrics.SeaTunnelMetricsContext;
import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
import org.apache.seatunnel.engine.server.service.slot.SlotService;
import org.apache.seatunnel.engine.server.task.CoordinatorTask;
@@ -51,6 +54,7 @@ import com.hazelcast.internal.serialization.Data;
import com.hazelcast.map.IMap;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -267,6 +271,37 @@ public class JobMasterTest extends
AbstractSeaTunnelServerTest {
() -> jobMaster.init(System.currentTimeMillis(), false));
}
+ @Test
+ void testFailedPipelineCleanupEnqueuesRecordAndRemovesMetrics() throws
Exception {
+ long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
+ JobMaster jobMaster = newJobInstanceWithRunningState(jobId);
+ PipelineLocation pipelineLocation =
getRunningPipelineLocation(jobMaster);
+
+ upsertMetricsForPipeline(pipelineLocation);
+ Assertions.assertTrue(hasMetricsForPipeline(pipelineLocation));
+
+ IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_PENDING_PIPELINE_CLEANUP);
+
+ try {
+ jobMaster.enqueuePipelineCleanupIfNeeded(pipelineLocation,
PipelineStatus.FAILED);
+
+ PipelineCleanupRecord cleanupRecord =
pendingCleanupIMap.get(pipelineLocation);
+ Assertions.assertNotNull(cleanupRecord);
+ Assertions.assertEquals(PipelineStatus.FAILED,
cleanupRecord.getFinalStatus());
+ Assertions.assertFalse(cleanupRecord.isSavepointEnd());
+
+ jobMaster.removeMetricsContext(pipelineLocation,
PipelineStatus.FAILED);
+
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
Assertions.assertFalse(hasMetricsForPipeline(pipelineLocation)));
+ } finally {
+ server.getCoordinatorService().cancelJob(jobId).join();
+ testIMapRemovedAfterJobComplete(jobId, jobMaster);
+ }
+ }
+
@Test
void testRestoreStartWithSavePointKeepsFinishedPipelines() throws
Exception {
long jobId =
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME).newId();
@@ -285,7 +320,9 @@ public class JobMasterTest extends
AbstractSeaTunnelServerTest {
private void assertCloseIdleTask(JobMaster jobMaster) {
SlotService slotService = server.getSlotService();
- Assertions.assertEquals(4,
slotService.getWorkerProfile().getAssignedSlots().length);
+ // Savepoint restore can overlap old slot release with new task
scheduling for a short time.
+ await().atMost(60, TimeUnit.SECONDS)
+ .until(() ->
slotService.getWorkerProfile().getAssignedSlots().length == 4);
Assertions.assertEquals(1,
jobMaster.getPhysicalPlan().getPipelineList().size());
SubPlan subPlan = jobMaster.getPhysicalPlan().getPipelineList().get(0);
@@ -395,4 +432,26 @@ public class JobMasterTest extends
AbstractSeaTunnelServerTest {
ConfigProvider.locateAndGetSeaTunnelConfig().getEngineConfig(),
server);
}
+
+ private PipelineLocation getRunningPipelineLocation(JobMaster jobMaster) {
+ return
jobMaster.getPhysicalPlan().getPipelineList().get(0).getPipelineLocation();
+ }
+
+ 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) {
+ return server.getEngineContext()
+ .getStateStores()
+ .metricsSnapshotStore()
+ .containsPipeline(pipelineLocation);
+ }
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecordTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecordTest.java
index b7f982a742..6586fd7549 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecordTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecordTest.java
@@ -84,8 +84,13 @@ class PipelineCleanupRecordTest {
Assertions.assertTrue(record.isCleaned());
}
+ /**
+ * Verifies that merge keeps cleanup progress while the newer terminal
status wins.
+ *
+ * <p>This protects later cleanup intents from being hidden behind stale
terminal statuses.
+ */
@Test
- void testMergeFromPrefersNonNullFieldsAndUnionsCollections() {
+ void testMergeFromUsesNewTerminalStatusAndUnionsCollections() {
PipelineLocation pipelineLocation1 = new PipelineLocation(1L, 1);
PipelineLocation pipelineLocation2 = new PipelineLocation(1L, 2);
TaskGroupLocation taskGroupLocation1 = new TaskGroupLocation(1L, 1,
1L);
@@ -128,7 +133,7 @@ class PipelineCleanupRecordTest {
PipelineCleanupRecord merged = record1.mergeFrom(record2);
Assertions.assertEquals(pipelineLocation1,
merged.getPipelineLocation());
- Assertions.assertEquals(PipelineStatus.FINISHED,
merged.getFinalStatus());
+ Assertions.assertEquals(PipelineStatus.CANCELED,
merged.getFinalStatus());
Assertions.assertTrue(merged.isSavepointEnd());
Assertions.assertTrue(merged.isMetricsImapCleaned());
