[FLINK-11380][tests] Finish mocking of resourceManagerClient before u…

…sing it

The problem was that we started the ResourceManager before we finished the mocking of the
resourceManagerClient. This lead then to concurrent modifications of the mock which made
the test fail.

flink-yarn/src/test/java/org/apache/flink/yarn/YarnFlinkResourceManagerTest.java

@@ -115,6 +115,9 @@ public void testYarnFlinkResourceManagerJobManagerLostLeadership() throws Except
  int numInitialTaskManagers = 5;
  final YarnResourceManagerCallbackHandler callbackHandler = new YarnResourceManagerCallbackHandler();
  AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = mock(AMRMClientAsync.class);
+ doReturn(Collections.singletonList(Collections.nCopies(numInitialTaskManagers, new AMRMClient.ContainerRequest(Resource.newInstance(1024 * 1024, 1), null, null, Priority.newInstance(0)))))
+ .when(resourceManagerClient).getMatchingRequests(any(Priority.class), anyString(), any(Resource.class));
+
  NMClient nodeManagerClient = mock(NMClient.class);
  UUID leaderSessionID = UUID.randomUUID();
 
@@ -181,9 +184,6 @@ public void testYarnFlinkResourceManagerJobManagerLostLeadership() throws Except
  nodeManagerClient
  ));
 
- doReturn(Collections.singletonList(Collections.nCopies(numInitialTaskManagers, new AMRMClient.ContainerRequest(Resource.newInstance(1024 * 1024, 1), null, null, Priority.newInstance(0)))))
- .when(resourceManagerClient).getMatchingRequests(any(Priority.class), anyString(), any(Resource.class));
-
  leaderRetrievalService.notifyListener(leader1.path().toString(), leaderSessionID);
 
  final AkkaActorGateway leader1Gateway = new AkkaActorGateway(leader1, leaderSessionID);