@GuardedBy("lock")
    private boolean isLeader(ResourceManager<?> resourceManager) {
        return running && this.leaderResourceManager == resourceManager;
    }
@Test
    void grantLeadership_withExistingLeader_stopExistLeader() throws Exception {
        final UUID leaderSessionId1 = UUID.randomUUID();
        final UUID leaderSessionId2 = UUID.randomUUID();
        final CompletableFuture<UUID> startRmFuture1 = new CompletableFuture<>();
        final CompletableFuture<UUID> startRmFuture2 = new CompletableFuture<>();
        final CompletableFuture<UUID> terminateRmFuture = new CompletableFuture<>();

        rmFactoryBuilder
                .setInitializeConsumer(
                        uuid -> {
                            if (!startRmFuture1.isDone()) {
                                startRmFuture1.complete(uuid);
                            } else {
                                startRmFuture2.complete(uuid);
                            }
                        })
                .setTerminateConsumer(terminateRmFuture::complete);

        createAndStartResourceManager();

        // first time grant leadership
        leaderElection.isLeader(leaderSessionId1).join();

        // second time grant leadership
        final CompletableFuture<LeaderInformation> confirmedLeaderInformation =
                leaderElection.isLeader(leaderSessionId2);

        // should terminate first RM, start a new RM and confirm leader session
        assertThatFuture(terminateRmFuture).eventuallySucceeds().isSameAs(leaderSessionId1);
        assertThatFuture(startRmFuture2).eventuallySucceeds().isSameAs(leaderSessionId2);
        assertThat(confirmedLeaderInformation.get().getLeaderSessionID())
                .isSameAs(leaderSessionId2);
    }