tillrohrmann commented on a change in pull request #14431: URL: https://github.com/apache/flink/pull/14431#discussion_r546659238
########## 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: yes will do. ---------------------------------------------------------------- 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: us...@infra.apache.org