@Override
    public boolean hasLeadership() {
        Preconditions.checkState(running.get());
        final Optional<KubernetesConfigMap> optionalConfigMap =
                kubeClient.getConfigMap(configMapName);

        if (optionalConfigMap.isPresent()) {
            return KubernetesLeaderElector.hasLeadership(optionalConfigMap.get(), lockIdentity);
        } else {
            leaderElectionListener.onError(
                    new KubernetesException(
                            String.format(
                                    "ConfigMap %s does not exist. This indicates that somebody has interfered with Flink's operation.",
                                    configMapName)));
            return false;
        }
    }
@Test
    void testHasLeadershipWhenConfigMapNotExist() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            leaderElectionDriver.hasLeadership();

                            final String expectedErrorMessage =
                                    "ConfigMap " + LEADER_CONFIGMAP_NAME + " does not exist.";

                            final LeaderElectionEvent.ErrorEvent errorEvent =
                                    electionEventHandler.await(
                                            LeaderElectionEvent.ErrorEvent.class);
                            assertThat(errorEvent.getError())
                                    .isInstanceOf(KubernetesException.class)
                                    .hasMessageContaining(expectedErrorMessage);
                        });
            }
        };
    }