Skip to content

Commit 6ee8f8d

Browse files
davidzollodybyteDanielLeens
authored
[Fix][Zeta] Clean failed pipeline metrics without full-map cleanup scan (#10757)
Co-authored-by: Doyeon Kim <132787602+dybyte@users.noreply.github.com> Co-authored-by: Daniel <escheduler@outlook.com> Co-authored-by: Daniel <48329107+DanielLeens@users.noreply.github.com> Co-authored-by: DanielLeens <DanielLeens@users.noreply.github.com>
1 parent 5b0f8a5 commit 6ee8f8d

8 files changed

Lines changed: 406 additions & 108 deletions

File tree

‎seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/config/EngineConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,7 @@ public void setJobMetricsBackupInterval(int jobMetricsBackupInterval) {
169169
public void setJobMetricsPartitionCount(int jobMetricsPartitionCount) {
170170
checkPositive(
171171
jobMetricsPartitionCount,
172-
ServerConfigOptions.MasterServerConfigOptions.JOB_METRICS_PARTITION_COUNT
172+
ServerConfigOptions.MasterServerConfigOptions.JOB_METRICS_PARTITION_COUNT.key()
173173
+ " must be > 0");
174174
this.jobMetricsPartitionCount = jobMetricsPartitionCount;
175175
}

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java‎

Lines changed: 124 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -602,71 +602,94 @@ private void cleanupPendingPipelines() {
602602
}
603603
}
604604

605-
private void processPendingPipelineCleanup(
605+
@VisibleForTesting
606+
void processPendingPipelineCleanup(
606607
PipelineLocation pipelineLocation, PipelineCleanupRecord record) {
607608
if (pipelineLocation == null || record == null) {
608609
return;
609610
}
610-
if (!shouldCleanup(record)) {
611-
removePendingCleanupRecord(pipelineLocation, record);
611+
IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
612+
pendingPipelineCleanupIMap;
613+
if (pendingCleanupIMap == null) {
612614
return;
613615
}
614616

615-
PipelineStatus currentStatus = getPipelineStatusFromIMap(pipelineLocation);
616-
if (currentStatus != null && !currentStatus.isEndState()) {
617-
return;
618-
}
617+
boolean locked = false;
618+
try {
619+
pendingCleanupIMap.lock(pipelineLocation);
620+
locked = true;
619621

620-
long now = System.currentTimeMillis();
621-
PipelineCleanupRecord updated = copy(record);
622-
updated.setLastAttemptTimeMillis(now);
623-
updated.setAttemptCount(record.getAttemptCount() + 1);
622+
// entrySet() is a snapshot. Revalidate it while holding the same key lock used by
623+
// restore invalidation so an old execution round cannot clean a newer round.
624+
PipelineCleanupRecord currentRecord = pendingCleanupIMap.get(pipelineLocation);
625+
if (!record.equals(currentRecord)) {
626+
return;
627+
}
628+
if (!shouldCleanup(record)) {
629+
removePendingCleanupRecord(pipelineLocation, record);
630+
return;
631+
}
624632

625-
if (!updated.isMetricsImapCleaned() && cleanupPipelineMetrics(pipelineLocation)) {
626-
updated.setMetricsImapCleaned(true);
627-
}
633+
PipelineStatus currentStatus = getPipelineStatusFromIMap(pipelineLocation);
634+
if (currentStatus != null && !currentStatus.isEndState()) {
635+
return;
636+
}
628637

629-
Map<TaskGroupLocation, Address> taskGroups = updated.getTaskGroups();
630-
if (taskGroups != null && !taskGroups.isEmpty()) {
631-
for (Map.Entry<TaskGroupLocation, Address> taskGroup : taskGroups.entrySet()) {
632-
TaskGroupLocation taskGroupLocation = taskGroup.getKey();
633-
if (updated.getCleanedTaskGroups() != null
634-
&& updated.getCleanedTaskGroups().contains(taskGroupLocation)) {
635-
continue;
636-
}
637-
Address workerAddress = taskGroup.getValue();
638-
if (workerAddress == null
639-
|| nodeEngine.getClusterService().getMember(workerAddress) == null) {
640-
continue;
641-
}
642-
try {
643-
NodeEngineUtil.sendOperationToMemberNode(
644-
nodeEngine,
645-
new CleanTaskGroupContextOperation(taskGroupLocation),
646-
workerAddress)
647-
.get();
648-
updated.getCleanedTaskGroups().add(taskGroupLocation);
649-
} catch (HazelcastInstanceNotActiveException e) {
650-
logger.warning(
651-
String.format(
652-
"%s clean TaskGroupContext failed: %s",
653-
taskGroupLocation, ExceptionUtils.getMessage(e)));
654-
} catch (Exception e) {
655-
logger.warning(
656-
String.format(
657-
"%s clean TaskGroupContext failed: %s",
658-
taskGroupLocation, ExceptionUtils.getMessage(e)),
659-
e);
638+
long now = System.currentTimeMillis();
639+
PipelineCleanupRecord updated = copy(record);
640+
updated.setLastAttemptTimeMillis(now);
641+
updated.setAttemptCount(record.getAttemptCount() + 1);
642+
643+
if (!updated.isMetricsImapCleaned() && cleanupPipelineMetrics(pipelineLocation)) {
644+
updated.setMetricsImapCleaned(true);
645+
}
646+
647+
Map<TaskGroupLocation, Address> taskGroups = updated.getTaskGroups();
648+
if (taskGroups != null && !taskGroups.isEmpty()) {
649+
for (Map.Entry<TaskGroupLocation, Address> taskGroup : taskGroups.entrySet()) {
650+
TaskGroupLocation taskGroupLocation = taskGroup.getKey();
651+
if (updated.getCleanedTaskGroups() != null
652+
&& updated.getCleanedTaskGroups().contains(taskGroupLocation)) {
653+
continue;
654+
}
655+
Address workerAddress = taskGroup.getValue();
656+
if (workerAddress == null
657+
|| nodeEngine.getClusterService().getMember(workerAddress) == null) {
658+
continue;
659+
}
660+
try {
661+
NodeEngineUtil.sendOperationToMemberNode(
662+
nodeEngine,
663+
new CleanTaskGroupContextOperation(taskGroupLocation),
664+
workerAddress)
665+
.get();
666+
updated.getCleanedTaskGroups().add(taskGroupLocation);
667+
} catch (HazelcastInstanceNotActiveException e) {
668+
logger.warning(
669+
String.format(
670+
"%s clean TaskGroupContext failed: %s",
671+
taskGroupLocation, ExceptionUtils.getMessage(e)));
672+
} catch (Exception e) {
673+
logger.warning(
674+
String.format(
675+
"%s clean TaskGroupContext failed: %s",
676+
taskGroupLocation, ExceptionUtils.getMessage(e)),
677+
e);
678+
}
660679
}
661680
}
662-
}
663681

664-
boolean replaced = pendingPipelineCleanupIMap.replace(pipelineLocation, record, updated);
665-
if (!replaced) {
666-
return;
667-
}
668-
if (updated.isCleaned()) {
669-
pendingPipelineCleanupIMap.remove(pipelineLocation, updated);
682+
boolean replaced = pendingCleanupIMap.replace(pipelineLocation, record, updated);
683+
if (!replaced) {
684+
return;
685+
}
686+
if (updated.isCleaned()) {
687+
pendingCleanupIMap.remove(pipelineLocation, updated);
688+
}
689+
} finally {
690+
if (locked) {
691+
pendingCleanupIMap.unlock(pipelineLocation);
692+
}
670693
}
671694
}
672695

@@ -690,7 +713,8 @@ private boolean shouldCleanup(PipelineCleanupRecord record) {
690713
if (record.isSavepointEnd()) {
691714
return false;
692715
}
693-
return PipelineStatus.CANCELED.equals(record.getFinalStatus())
716+
return PipelineStatus.FAILED.equals(record.getFinalStatus())
717+
|| PipelineStatus.CANCELED.equals(record.getFinalStatus())
694718
|| PipelineStatus.FINISHED.equals(record.getFinalStatus());
695719
}
696720

@@ -1054,6 +1078,50 @@ private void cleanupPendingJobStateForRestore(long jobId, JobCleanupRecord recor
10541078
runningJobInfoIMap.remove(jobId);
10551079
}
10561080

1081+
/**
1082+
* Removes pending pipeline cleanup records from an older execution round of the same job.
1083+
*
1084+
* <p>A savepoint-start submission reuses the same job id and pipeline ids, while pipeline
1085+
* cleanup records are keyed only by {@link PipelineLocation}. If a previous FAILED record
1086+
* survives into the restored round, it can delete metrics that the later savepoint-end round
1087+
* must retain.
1088+
*
1089+
* @param jobId job id that is entering a new savepoint-start execution round
1090+
*/
1091+
@VisibleForTesting
1092+
void cleanupPendingPipelineCleanupForRestore(long jobId) {
1093+
IMap<PipelineLocation, PipelineCleanupRecord> pendingCleanupIMap =
1094+
pendingPipelineCleanupIMap;
1095+
if (pendingCleanupIMap == null || pendingCleanupIMap.isEmpty()) {
1096+
return;
1097+
}
1098+
for (Map.Entry<PipelineLocation, PipelineCleanupRecord> entry :
1099+
pendingCleanupIMap.entrySet()) {
1100+
PipelineLocation pipelineLocation = entry.getKey();
1101+
if (pipelineLocation == null) {
1102+
continue;
1103+
}
1104+
boolean locked = false;
1105+
try {
1106+
pendingCleanupIMap.lock(pipelineLocation);
1107+
locked = true;
1108+
PipelineCleanupRecord currentRecord = pendingCleanupIMap.get(pipelineLocation);
1109+
boolean sameJob =
1110+
pipelineLocation.getJobId() == jobId
1111+
|| (currentRecord != null
1112+
&& currentRecord.getPipelineLocation() != null
1113+
&& currentRecord.getPipelineLocation().getJobId() == jobId);
1114+
if (sameJob) {
1115+
pendingCleanupIMap.remove(pipelineLocation);
1116+
}
1117+
} finally {
1118+
if (locked) {
1119+
pendingCleanupIMap.unlock(pipelineLocation);
1120+
}
1121+
}
1122+
}
1123+
}
1124+
10571125
/**
10581126
* Polls master ownership and reconciles the local coordinator lifecycle with cluster state.
10591127
*
@@ -1201,6 +1269,9 @@ public PassiveCompletableFuture<Void> submitJob(
12011269
JobMaster jobMaster = null;
12021270
JobInfo submittedJobInfo = null;
12031271
try {
1272+
if (isStartWithSavePoint) {
1273+
cleanupPendingPipelineCleanupForRestore(jobId);
1274+
}
12041275
JobCleanupRecord pendingCleanupRecord =
12051276
pendingJobCleanupIMap != null
12061277
? pendingJobCleanupIMap.get(jobId)

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/common/statestore/metrics/hazelcast/HazelcastMetricsSnapshotStateStore.java‎

Lines changed: 17 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -30,14 +30,12 @@
3030

3131
import java.util.HashMap;
3232
import java.util.HashSet;
33-
import java.util.List;
3433
import java.util.Map;
3534
import java.util.Set;
3635
import java.util.UUID;
3736
import java.util.concurrent.ConcurrentHashMap;
3837
import java.util.concurrent.atomic.AtomicBoolean;
3938
import java.util.concurrent.atomic.AtomicLong;
40-
import java.util.stream.Collectors;
4139

4240
/** Implementation backed by a partitioned Hazelcast metrics {@link IMap}. */
4341
public class HazelcastMetricsSnapshotStateStore
@@ -115,40 +113,24 @@ public void remove(final TaskLocation taskLocation) {
115113

116114
@Override
117115
public void removePipeline(final PipelineLocation pipelineLocation) {
118-
Map<Long, List<TaskLocation>> partitionedTasks = new HashMap<>();
119-
for (Map.Entry<Long, Map<TaskLocation, SeaTunnelMetricsContext>> entry :
120-
metricsImap.entrySet()) {
121-
long partition = entry.getKey();
122-
List<TaskLocation> tasksToRemove =
123-
entry.getValue().keySet().stream()
124-
.filter(
125-
t ->
126-
t.getTaskGroupLocation()
127-
.getPipelineLocation()
128-
.equals(pipelineLocation))
129-
.collect(Collectors.toList());
130-
if (!tasksToRemove.isEmpty()) {
131-
partitionedTasks.put(partition, tasksToRemove);
132-
}
116+
for (long partition = 0; partition < partitionCount; partition++) {
117+
metricsImap.compute(
118+
partition,
119+
(ignored, current) -> {
120+
if (current == null || current.isEmpty()) {
121+
return current;
122+
}
123+
Map<TaskLocation, SeaTunnelMetricsContext> updated = new HashMap<>(current);
124+
updated.entrySet()
125+
.removeIf(
126+
entry ->
127+
pipelineLocation.equals(
128+
entry.getKey()
129+
.getTaskGroupLocation()
130+
.getPipelineLocation()));
131+
return updated.isEmpty() ? null : updated;
132+
});
133133
}
134-
135-
partitionedTasks
136-
.entrySet()
137-
.parallelStream()
138-
.forEach(
139-
entry -> {
140-
long partition = entry.getKey();
141-
List<TaskLocation> tasks = entry.getValue();
142-
metricsImap.compute(
143-
partition,
144-
(k, oldVal) -> {
145-
if (oldVal != null) {
146-
tasks.forEach(oldVal::remove);
147-
if (oldVal.isEmpty()) return null;
148-
}
149-
return oldVal;
150-
});
151-
});
152134
}
153135

154136
@Override

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1114,8 +1114,11 @@ public void enqueuePipelineCleanupIfNeeded(
11141114
PipelineStatus.FINISHED.equals(pipelineStatus)
11151115
&& checkpointManager != null
11161116
&& checkpointManager.isPipelineSavePointEnd(pipelineLocation);
1117+
// Failed pipelines also need cleanup so their distributed metrics do not leak into later
1118+
// task recovery or re-submission flows.
11171119
boolean shouldCleanup =
1118-
PipelineStatus.CANCELED.equals(pipelineStatus)
1120+
PipelineStatus.FAILED.equals(pipelineStatus)
1121+
|| PipelineStatus.CANCELED.equals(pipelineStatus)
11191122
|| (PipelineStatus.FINISHED.equals(pipelineStatus) && !savepointEnd);
11201123
if (!shouldCleanup) {
11211124
return;
@@ -1172,7 +1175,8 @@ public void enqueuePipelineCleanupIfNeeded(
11721175

11731176
public void removeMetricsContext(
11741177
PipelineLocation pipelineLocation, PipelineStatus pipelineStatus) {
1175-
if ((pipelineStatus.equals(PipelineStatus.FINISHED)
1178+
if (pipelineStatus.equals(PipelineStatus.FAILED)
1179+
|| (pipelineStatus.equals(PipelineStatus.FINISHED)
11761180
&& !checkpointManager.isPipelineSavePointEnd(pipelineLocation))
11771181
|| pipelineStatus.equals(PipelineStatus.CANCELED)) {
11781182

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/cleanup/PipelineCleanupRecord.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,16 @@ public boolean isCleaned() {
136136
&& cleanedTaskGroups.containsAll(taskGroups.keySet());
137137
}
138138

139+
/**
140+
* Merges a newer cleanup intent into this record while preserving completed cleanup progress.
141+
*
142+
* <p>The caller passes the newer terminal-state record as {@code other}. Its final status must
143+
* win so an older FAILED or CANCELED intent cannot keep driving cleanup after a later terminal
144+
* state is reported for the same pipeline location.
145+
*
146+
* @param other newer cleanup record for the same pipeline location
147+
* @return merged cleanup record
148+
*/
139149
public PipelineCleanupRecord mergeFrom(PipelineCleanupRecord other) {
140150
if (other == null) {
141151
return this;
@@ -161,7 +171,7 @@ public PipelineCleanupRecord mergeFrom(PipelineCleanupRecord other) {
161171
this.pipelineLocation != null
162172
? this.pipelineLocation
163173
: other.pipelineLocation,
164-
this.finalStatus != null ? this.finalStatus : other.finalStatus,
174+
other.finalStatus != null ? other.finalStatus : this.finalStatus,
165175
this.savepointEnd || other.savepointEnd,
166176
mergedTaskGroups,
167177
mergedCleaned,

0 commit comments

Comments
 (0)