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 dc032347a1 [Fix][Zeta] Fix flaky pending job info test (#10891)
dc032347a1 is described below
commit dc032347a15aa58410c140394751b60518810f3d
Author: QuakeWang <[email protected]>
AuthorDate: Sat May 30 13:45:13 2026 +0800
[Fix][Zeta] Fix flaky pending job info test (#10891)
Signed-off-by: QuakeWang <[email protected]>
---
.../engine/server/CoordinatorServiceTest.java | 34 +++++++++++++---------
1 file changed, 21 insertions(+), 13 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
index a247c87f43..c626432f69 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
@@ -44,6 +44,7 @@ import
org.apache.seatunnel.engine.server.execution.PendingSourceState;
import org.apache.seatunnel.engine.server.execution.TaskGroupContext;
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.operation.PrintMessageOperation;
@@ -1034,22 +1035,29 @@ public class CoordinatorServiceTest {
@Test
void testGetPendingJobInfo() {
- JobInformation jobInformation =
- submitJob(
- "CoordinatorServiceTest_testGetPendingJobInfo",
- "batch_fake_to_console.conf",
- "test_get_pending_job_info");
-
- CoordinatorService coordinatorService =
jobInformation.coordinatorService;
- Long jobId = jobInformation.jobId;
+ SeaTunnelServer server = Mockito.mock(SeaTunnelServer.class);
+ CoordinatorService coordinatorService =
newMockCoordinatorService(server);
+ try {
+ Long jobId = 70001L;
+ JobHistoryService jobHistoryService =
Mockito.mock(JobHistoryService.class);
+ ReflectionUtils.setField(coordinatorService, "jobHistoryService",
jobHistoryService);
+
Mockito.when(jobHistoryService.getJobDAGInfo(jobId)).thenReturn(null);
-
Assertions.assertTrue(coordinatorService.getPendingJobQueue().contains(jobId));
+ JobDAGInfo pendingJobDAGInfo = new JobDAGInfo();
+ pendingJobDAGInfo.setJobId(jobId);
+ JobMaster jobMaster =
+ enqueueMockPendingJob(coordinatorService, jobId, new
CountDownLatch(1));
+
Mockito.when(jobMaster.getJobDAGInfo()).thenReturn(pendingJobDAGInfo);
- JobDAGInfo jobDAGInfo =
- Assertions.assertDoesNotThrow(() ->
coordinatorService.getJobInfo(jobId));
- Assertions.assertEquals(jobId, jobDAGInfo.getJobId());
+
Assertions.assertTrue(coordinatorService.getPendingJobQueue().contains(jobId));
- jobInformation.coordinatorServiceTest.shutdown();
+ JobDAGInfo jobDAGInfo =
+ Assertions.assertDoesNotThrow(() ->
coordinatorService.getJobInfo(jobId));
+ Assertions.assertSame(pendingJobDAGInfo, jobDAGInfo);
+ Assertions.assertEquals(jobId, jobDAGInfo.getJobId());
+ } finally {
+ coordinatorService.shutdown();
+ }
}
@Test
