phet commented on code in PR #4027:
URL: https://github.com/apache/gobblin/pull/4027#discussion_r1719075016
##########
gobblin-service/src/test/java/org/apache/gobblin/service/modules/orchestration/proc/KillDagProcTest.java:
##########
@@ -112,15 +141,35 @@ public void killDag() throws IOException,
URISyntaxException, InterruptedExcepti
null, this.dagManagementStateStore, mockedDagProcEngineMetrics),
ConfigFactory.empty());
killDagProc.process(this.dagManagementStateStore,
this.mockedDagProcEngineMetrics);
- long cancelJobCount = specProducers.stream()
+ int numOfLaunchedJobs = 1;
+ int numOfCancelledJobs = 5; // all jobs in the dag
+ int numOfCancelledFlows = 1;
+ int numOfCancelledJobsWithJobFuture = numOfLaunchedJobs;
+ long actualCancelJobCount = specProducers.stream()
.mapToLong(p -> Mockito.mockingDetails(p)
.getInvocations()
.stream()
.filter(a -> a.getMethod().getName().equals("cancelJob"))
.count())
.sum();
+
+ // kill dag procs kill only the launched jobs with parameters containing
jobFuture
+ Mockito.verify(specProducers.get(0),
Mockito.times(numOfCancelledJobsWithJobFuture)).cancelJob(any(), argThat(props
->
+ props.getProperty(ConfigurationKeys.SPEC_PRODUCER_SERIALIZED_FUTURE,
"ABSENT").equals(MockedSpecExecutor.dummySerializedFuture)));
+
+ // job future object is not available for rest of the jobs cancel
parameters
+ specProducers.stream()
+ .skip(numOfCancelledJobsWithJobFuture) // separately verified
`specProducers.get(0)` above
+ .forEach(sp -> Mockito.verify(sp, Mockito.never()).cancelJob(any(),
argThat(props ->
+
props.getProperty(ConfigurationKeys.SPEC_PRODUCER_SERIALIZED_FUTURE,
"ABSENT").equals(MockedSpecExecutor.dummySerializedFuture))));
+
// kill dag proc tries to cancel all the dag nodes
- Assert.assertEquals(cancelJobCount, 5);
+ Assert.assertEquals(actualCancelJobCount, numOfCancelledJobs);
+
+ Mockito.verify(this.mockedEventSubmitter,
Mockito.times(numOfCancelledJobs))
+ .submit(eq(TimingEvent.LauncherTimings.JOB_CANCEL), anyMap());
+ Mockito.verify(this.mockedEventSubmitter,
Mockito.times(numOfCancelledFlows))
+ .submit(eq(TimingEvent.FlowTimings.FLOW_CANCELLED), anyMap());
Review Comment:
nice work!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
