zentol commented on a change in pull request #14431:
URL: https://github.com/apache/flink/pull/14431#discussion_r546605468
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
##########
@@ -797,52 +796,75 @@ private void startJobExecution() throws Exception {
}
private void startJobMasterServices() throws Exception {
- // start the slot pool make sure the slot pool now accepts
messages for this leader
- slotPool.start(getFencingToken(), getAddress(),
getMainThreadExecutor());
-
- //TODO: Remove once the ZooKeeperLeaderRetrieval returns the
stored address upon start
- // try to reconnect to previously known leader
- reconnectToResourceManager(new FlinkException("Starting
JobMaster component."));
-
- // job is ready to go, try to establish connection with
resource manager
- // - activate leader retrieval for the resource manager
- // - on notification of the leader, the connection will be
established and
- // the slot pool will start requesting slots
- resourceManagerLeaderRetriever.start(new
ResourceManagerLeaderListener());
+ try {
+ // start the slot pool make sure the slot pool now
accepts messages for this leader
+ slotPool.start(getFencingToken(), getAddress(),
getMainThreadExecutor());
+
+ // job is ready to go, try to establish connection with
resource manager
+ // - activate leader retrieval for the resource
manager
+ // - on notification of the leader, the connection
will be established and
+ // the slot pool will start requesting slots
+ resourceManagerLeaderRetriever.start(new
ResourceManagerLeaderListener());
+ } catch (Exception e) {
+ handleStartJobMasterServicesError(e);
+ }
}
- /**
- * Suspending job, all the running tasks will be cancelled, and
communication with other components
- * will be disposed.
- *
- * @param cause The reason of why this job been suspended.
- */
- private void suspendExecution(final Exception cause) {
- validateRunsInMainThread();
+ private void handleStartJobMasterServicesError(Exception e) throws
Exception {
+ try {
+ stopJobMasterServices();
+ } catch (Exception inner) {
+ e.addSuppressed(inner);
+ }
+
+ throw e;
+ }
+
+ private void stopJobMasterServices() throws Exception {
+ Exception resultingException = null;
try {
resourceManagerLeaderRetriever.stop();
- resourceManagerAddress = null;
- } catch (Throwable t) {
- log.warn("Failed to stop resource manager leader
retriever when suspending.", t);
+ } catch (Exception e) {
+ resultingException = e;
}
- suspendScheduler(cause);
+ // TODO: Distinguish between job termination which should free
all slots and a loss of leadership which should keep the slots
+ slotPool.close();
+
+ stopHeartbeatServices();
+
+ ExceptionUtils.tryRethrowException(resultingException);
+ }
+
+ private void stopJobExecution(final Exception cause) throws Exception {
+ validateRunsInMainThread();
+
+ Exception resultingException = null;
Review comment:
Move this closer to the try block?
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterSchedulerTest.java
##########
@@ -72,16 +70,13 @@ public void
testIfStartSchedulingFailsJobMasterFailsFatally() throws Exception {
System.setSecurityManager(trackingSecurityManager);
final SchedulerNGFactory schedulerFactory = new
FailingSchedulerFactory();
- final JobMaster jobMaster = new JobMasterBuilder(new
JobGraph(), TESTING_RPC_SERVICE_RESOURCE
- .getTestingRpcService())
+ final JobMaster jobMaster = new JobMasterBuilder(new
JobGraph(), TESTING_RPC_SERVICE_RESOURCE.getTestingRpcService())
.withSchedulerFactory(schedulerFactory)
.createJobMaster();
- final CompletableFuture<Acknowledge> startFuture =
jobMaster.start(JobMasterId.generate());
+ jobMaster.start();
try {
- startFuture.join();
-
assertThat(trackingSecurityManager.getSystemExitFuture().join(),
is(FatalExitExceptionHandler.EXIT_CODE));
Review comment:
This test got stuck on CI; I suppose we never call the fatal error
handler since an error during the start of the scheduling now prevents the
RpcEndpoint from starting altogether.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
