Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5107#discussion_r154748301
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
---
@@ -244,7 +302,32 @@ protected JobManagerRunner createJobManagerRunner(
FatalErrorHandler fatalErrorHandler) throws
Exception {
assertEquals(expectedJobId, jobGraph.getJobID());
- return jobManagerRunner;
+ return new JobManagerRunner(resourceId, jobGraph,
configuration, rpcService,
+ highAvailabilityServices, heartbeatServices,
jobManagerServices, metricRegistry,
+ onCompleteActions, fatalErrorHandler);
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> submitJob(final JobGraph
jobGraph, final Time timeout) {
+ final CompletableFuture<Acknowledge> submitJobFuture =
super.submitJob(jobGraph, timeout);
+
+ try {
+ submitJobFuture.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ submitJobLatch.countDown();
+ return submitJobFuture;
+ }
+
+ @Override
+ void recoverJobs() {
+ if (recoverJobsEnabled.get()) {
--- End diff --
Without this I do not see how I can verify whether a job was submitted
regularly or via `recoverJobs`.
---
