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 54a176c5de [Fix][Zeta] Make finished jobs API tolerate missing history
data (#11269)
54a176c5de is described below
commit 54a176c5decf97a480293ca190e76bc13326b617
Author: Jast <[email protected]>
AuthorDate: Tue Jul 7 23:48:54 2026 +0800
[Fix][Zeta] Make finished jobs API tolerate missing history data (#11269)
---
.../engine/server/rest/service/JobInfoService.java | 50 +++++++++++++---------
.../rest/service/JobInfoServiceNullSafetyTest.java | 27 ++++++++++++
2 files changed, 57 insertions(+), 20 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java
index c54352afac..13e243f9fb 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java
@@ -32,7 +32,6 @@ import
org.apache.seatunnel.engine.core.job.JobImmutableInformation;
import org.apache.seatunnel.engine.core.job.JobInfo;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.master.JobHistoryService.JobState;
-import org.apache.seatunnel.engine.server.operation.GetJobMetricsOperation;
import org.apache.seatunnel.engine.server.operation.GetJobStatusOperation;
import org.apache.seatunnel.engine.server.rest.ConfigFormat;
import org.apache.seatunnel.engine.server.rest.RestConstant;
@@ -121,9 +120,8 @@ public class JobInfoService extends BaseService {
IMap<Long, JobDAGInfo> finishedJobDAGInfo =
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_FINISHED_JOB_VERTEX_INFO);
- SeaTunnelServer seaTunnelServer = getSeaTunnelServer(true);
-
return finishedJob.values().stream()
+ .filter(java.util.Objects::nonNull)
.filter(
jobState -> {
if (state.isEmpty()) {
@@ -131,31 +129,43 @@ public class JobInfoService extends BaseService {
}
return
jobState.getJobStatus().name().equals(state.toUpperCase());
})
- .sorted(Comparator.comparing(JobState::getFinishTime,
Comparator.reverseOrder()))
+ .sorted(
+ Comparator.comparing(
+ JobState::getFinishTime,
+
Comparator.nullsLast(Comparator.reverseOrder())))
.map(
jobState -> {
Long jobId = jobState.getJobId();
- String jobMetrics;
- if (seaTunnelServer == null) {
- jobMetrics =
- (String)
-
NodeEngineUtil.sendOperationToMasterNode(
- nodeEngine,
- new
GetJobMetricsOperation(jobId))
- .join();
- } else {
- jobMetrics =
- seaTunnelServer
- .getCoordinatorService()
- .getJobMetrics(jobId)
- .toJsonString();
- }
return getJobInfoJson(
- jobState, jobMetrics,
finishedJobDAGInfo.get(jobId));
+ jobState,
+ getFinishedJobMetricsJson(jobId),
+ getFinishedJobDAGInfo(finishedJobDAGInfo,
jobId));
})
.collect(JsonArray::new, JsonArray::add, JsonArray::add);
}
+ private String getFinishedJobMetricsJson(Long jobId) {
+ try {
+ IMap<Long, JobMetrics> finishedJobMetrics =
+
nodeEngine.getHazelcastInstance().getMap(Constant.IMAP_FINISHED_JOB_METRICS);
+ JobMetrics metrics = finishedJobMetrics.getOrDefault(jobId,
JobMetrics.empty());
+ return metrics == null ? JobMetrics.empty().toJsonString() :
metrics.toJsonString();
+ } catch (Throwable t) {
+ log.warn("Failed to load finished job metrics for job {}: {}",
jobId, t.getMessage());
+ return JobMetrics.empty().toJsonString();
+ }
+ }
+
+ private JobDAGInfo getFinishedJobDAGInfo(
+ IMap<Long, JobDAGInfo> finishedJobDAGInfo, Long jobId) {
+ try {
+ return finishedJobDAGInfo.get(jobId);
+ } catch (Throwable t) {
+ log.warn("Failed to load finished job DAG for job {}: {}", jobId,
t.getMessage());
+ return null;
+ }
+ }
+
public JsonArray getRunningJobsJson() {
IMap<Long, JobInfo> values = getRunningJobInfoMap();
SeaTunnelServer seaTunnelServer = getSeaTunnelServer(true);
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/JobInfoServiceNullSafetyTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/JobInfoServiceNullSafetyTest.java
index 19ce8ce314..151f60b4ed 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/JobInfoServiceNullSafetyTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/JobInfoServiceNullSafetyTest.java
@@ -29,10 +29,12 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.internal.json.JsonArray;
import com.hazelcast.internal.json.JsonObject;
import com.hazelcast.map.IMap;
import com.hazelcast.spi.impl.NodeEngineImpl;
+import java.util.Arrays;
import java.util.Collections;
import static org.mockito.Mockito.mock;
@@ -134,4 +136,29 @@ class JobInfoServiceNullSafetyTest {
Assertions.assertNotNull(result);
Assertions.assertEquals(jobId.toString(),
result.getString(RestConstant.JOB_ID, null));
}
+
+ @Test
+ void shouldListFinishedJobsWhenFinishTimeMetricsOrDagAreMissing() {
+ Long jobWithFinishTime = 2L;
+ JobHistoryService.JobState missingFinishTime = buildJobState(jobId,
1000L, null);
+ JobHistoryService.JobState completed =
buildJobState(jobWithFinishTime, 1000L, 2000L);
+
+ when(finishedJobStateMap.values())
+ .thenReturn((java.util.Collection)
Arrays.asList(missingFinishTime, completed));
+ when(finishedJobMetricsMap.getOrDefault(jobId,
JobMetrics.empty())).thenReturn(null);
+ when(finishedJobMetricsMap.getOrDefault(jobWithFinishTime,
JobMetrics.empty()))
+ .thenThrow(new RuntimeException("metrics unavailable"));
+ when(finishedJobVertexInfoMap.get(jobId))
+ .thenThrow(new RuntimeException("dag unavailable"));
+ when(finishedJobVertexInfoMap.get(jobWithFinishTime)).thenReturn(null);
+
+ JsonArray result = jobInfoService.getJobsByStateJson("");
+
+ Assertions.assertEquals(2, result.size());
+ Assertions.assertEquals(
+ jobWithFinishTime.toString(),
+ result.get(0).asObject().getString(RestConstant.JOB_ID, null));
+ Assertions.assertEquals(
+ jobId.toString(),
result.get(1).asObject().getString(RestConstant.JOB_ID, null));
+ }
}
