@Override
    public void publishLeaderInformation(String componentId, LeaderInformation leaderInformation) {
        Preconditions.checkState(running.get());

        try {
            kubeClient
                    .checkAndUpdateConfigMap(
                            configMapName,
                            updateConfigMapWithLeaderInformation(componentId, leaderInformation))
                    .get();
        } catch (InterruptedException | ExecutionException e) {
            leaderElectionListener.onError(e);
        }

        LOG.debug(
                "Successfully wrote leader information {} for leader {} into the config map {}.",
                leaderInformation,
                componentId,
                configMapName);
    }
@Test
    void testLeaderConfigMapModifiedExternallyShouldBeCorrected() throws Exception {
        new Context() {
            {
                runTest(
                        () -> {
                            final UUID leaderSessionID = leaderCallbackGrantLeadership();

                            final Optional<LeaderInformation> leaderInformation =
                                    getLeaderInformationFromConfigMap();
                            assertThat(leaderInformation.map(LeaderInformation::getLeaderSessionID))
                                    .hasValue(leaderSessionID);

                            // Update ConfigMap with wrong data
                            final UUID faultySessionID = UUID.randomUUID();
                            final String faultyAddress = "faultyLeaderAddress";
                            putLeaderInformationIntoConfigMap(faultySessionID, faultyAddress);

                            // Simulate ConfigMap modification event
                            final FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap>
                                    callbackHandler = getLeaderElectionConfigMapCallback();
                            callbackHandler.onModified(
                                    Collections.singletonList(getLeaderConfigMap()));

                            final LeaderInformationRegister leaderInformationRegister =
                                    electionEventHandler
                                            .await(
                                                    LeaderElectionEvent
                                                            .AllLeaderInformationChangeEvent.class)
                                            .getLeaderInformationRegister();

                            assertThat(leaderInformationRegister.getRegisteredComponentIds())
                                    .containsExactly(componentId);

                            final LeaderInformation expectedFaultyLeaderInformation =
                                    LeaderInformation.known(faultySessionID, faultyAddress);
                            assertThat(leaderInformationRegister.forComponentId(componentId))
                                    .hasValue(expectedFaultyLeaderInformation);

                            leaderElectionDriver.publishLeaderInformation(
                                    componentId,
                                    LeaderInformation.known(leaderSessionID, leaderAddress));

                            assertThat(getLeaderInformationFromConfigMap())
                                    .as("LeaderInformation should have been corrected.")
                                    .hasValue(
                                            LeaderInformation.known(
                                                    leaderSessionID, leaderAddress));
                        });
            }
        };
    }