@Override
    public void start() throws Exception {
        LOG.debug("Start leadership runner for job {}.", getJobID());
        leaderElection.startLeaderElection(this);
    }
@Test
    void testJobMasterServiceLeadershipRunnerCloseWhenElectionServiceGrantLeaderShip()
            throws Exception {
        final AtomicReference<LeaderInformationRegister> storedLeaderInformation =
                new AtomicReference<>(LeaderInformationRegister.empty());
        final AtomicBoolean haBackendLeadershipFlag = new AtomicBoolean();

        final TestingLeaderElectionDriver.Factory driverFactory =
                new TestingLeaderElectionDriver.Factory(
                        TestingLeaderElectionDriver.newBuilder(
                                haBackendLeadershipFlag,
                                storedLeaderInformation,
                                new AtomicBoolean()));

        // we need to use DefaultLeaderElectionService here because JobMasterServiceLeadershipRunner
        // in connection with the DefaultLeaderElectionService generates the nested locking
        final DefaultLeaderElectionService defaultLeaderElectionService =
                new DefaultLeaderElectionService(driverFactory, fatalErrorHandler);

        // latch to detect when we reached the first synchronized section having a lock on the
        // JobMasterServiceProcess#stop side
        final OneShotLatch closeAsyncCalledTrigger = new OneShotLatch();
        // latch to halt the JobMasterServiceProcess#stop before calling stop on the
        // DefaultLeaderElectionService instance (and entering the LeaderElectionService's
        // synchronized block)
        final OneShotLatch triggerClassLoaderLeaseRelease = new OneShotLatch();

        final JobMasterServiceProcess jobMasterServiceProcess =
                TestingJobMasterServiceProcess.newBuilder()
                        .setGetJobMasterGatewayFutureSupplier(CompletableFuture::new)
                        .setGetResultFutureSupplier(CompletableFuture::new)
                        .setGetLeaderAddressFutureSupplier(
                                () -> CompletableFuture.completedFuture("unused address"))
                        .setCloseAsyncSupplier(
                                () -> {
                                    closeAsyncCalledTrigger.trigger();
                                    // we have to return a completed future because we need the
                                    // follow-up task to run in the calling thread to make the
                                    // follow-up code block being executed in the synchronized block
                                    return CompletableFuture.completedFuture(null);
                                })
                        .build();
        final String componentId = "random-component-id";
        final LeaderElection leaderElection =
                defaultLeaderElectionService.createLeaderElection(componentId);
        try (final JobMasterServiceLeadershipRunner jobManagerRunner =
                newJobMasterServiceLeadershipRunnerBuilder()
                        .setClassLoaderLease(
                                TestingClassLoaderLease.newBuilder()
                                        .setCloseRunnable(
                                                () -> {
                                                    try {
                                                        // we want to wait with releasing to halt
                                                        // before calling stop on the
                                                        // DefaultLeaderElectionService
                                                        triggerClassLoaderLeaseRelease.await();

                                                        // In order to reproduce the deadlock, we
                                                        // need to ensure that
                                                        // leaderContender#grantLeadership can be
                                                        // called after
                                                        // JobMasterServiceLeadershipRunner obtains
                                                        // its own lock. Unfortunately, This will
                                                        // change the running status of
                                                        // DefaultLeaderElectionService
                                                        // to false, which will cause the
                                                        // notification of leadership to be
                                                        // ignored. The issue is that we
                                                        // don't have any means of verify that we're
                                                        // in the synchronized block of
                                                        // DefaultLeaderElectionService#lock in
                                                        // DefaultLeaderElectionService#onGrantLeadership,
                                                        // but we trigger this implicitly through
                                                        // TestingLeaderElectionDriver#grantLeadership(UUID).
                                                        // Adding a short sleep can ensure that
                                                        // another thread successfully receives the
                                                        // leadership notification, so that the
                                                        // deadlock problem can recur.
                                                        Thread.sleep(5);
                                                    } catch (InterruptedException e) {
                                                        ExceptionUtils.checkInterrupted(e);
                                                    }
                                                })
                                        .build())
                        .setJobMasterServiceProcessFactory(
                                TestingJobMasterServiceProcessFactory.newBuilder()
                                        .setJobMasterServiceProcessFunction(
                                                ignoredSessionId -> jobMasterServiceProcess)
                                        .build())
                        .setLeaderElection(leaderElection)
                        .build()) {
            jobManagerRunner.start();

            // grant leadership to create jobMasterServiceProcess
            haBackendLeadershipFlag.set(true);
            final UUID leaderSessionID = UUID.randomUUID();
            defaultLeaderElectionService.onGrantLeadership(leaderSessionID);

            final SupplierWithException<Boolean, Exception> confirmationForSessionIdReceived =
                    () ->
                            storedLeaderInformation
                                    .get()
                                    .forComponentId(componentId)
                                    .map(LeaderInformation::getLeaderSessionID)
                                    .map(sessionId -> sessionId.equals(leaderSessionID))
                                    .orElse(false);
            CommonTestUtils.waitUntilCondition(confirmationForSessionIdReceived);

            final CheckedThread contenderCloseThread = createCheckedThread(jobManagerRunner::close);
            contenderCloseThread.start();

            // waiting for the contender reaching the synchronized section of the stop call
            closeAsyncCalledTrigger.await();

            final CheckedThread grantLeadershipThread =
                    createCheckedThread(
                            () -> {
                                // DefaultLeaderElectionService enforces a proper event handling
                                // order (i.e. no two grant or revoke events should appear after
                                // each other). This requires the leadership to be revoked before
                                // regaining leadership in this test.
                                defaultLeaderElectionService.onRevokeLeadership();
                                defaultLeaderElectionService.onGrantLeadership(UUID.randomUUID());
                            });
            grantLeadershipThread.start();

            // finalize ClassloaderLease release to trigger DefaultLeaderElectionService#stop
            triggerClassLoaderLeaseRelease.trigger();

            contenderCloseThread.sync();
            grantLeadershipThread.sync();
        }
    }