[GitHub] [flink] tillrohrmann commented on a change in pull request #13644: [FLINK-19542][k8s]Implement LeaderElectionService and LeaderRetrievalService based on Kubernetes API

[GitBox <gi...@apache.org>]

tillrohrmann commented on a change in pull request #13644:
URL: https://github.com/apache/flink/pull/13644#discussion_r508569365



##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	/** The executor to run external IO operations on. */
+	protected final Executor executor;
+
+	/** The runtime configuration. */
+	protected final Configuration configuration;
+
+	/** Store for arbitrary blobs. */
+	private final BlobStoreService blobStoreService;
+
+	public AbstractHaServices(Executor executor, Configuration config, BlobStoreService blobStoreService) {

Review comment:
       nit: I try to always pass `Configuration` first because it represents the part which configures the component.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).

Review comment:
       Maybe add a bit better description what this class implements (e.g. if I want to implement `HighAvailabilityServices` what does this class already solve for me?).

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/AbstractLeaderElectionService.java
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.leaderelection;
+
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.UUID;
+
+/**
+ * Abstract class for leader election service based on distributed coordination system(e.g. Zookeeper, Kubernetes, etc.).
+ */
+public abstract class AbstractLeaderElectionService implements LeaderElectionService {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	protected final Object lock = new Object();
+
+	/** The leader contender which applies for leadership. */
+	@GuardedBy("lock")
+	protected volatile LeaderContender leaderContender;
+
+	@GuardedBy("lock")
+	private volatile UUID issuedLeaderSessionID;
+
+	@GuardedBy("lock")
+	protected volatile UUID confirmedLeaderSessionID;
+
+	@GuardedBy("lock")
+	protected volatile String confirmedLeaderAddress;
+
+	@GuardedBy("lock")
+	protected volatile boolean running;
+
+	protected AbstractLeaderElectionService() {
+		leaderContender = null;
+
+		issuedLeaderSessionID = null;
+		confirmedLeaderSessionID = null;
+		confirmedLeaderAddress = null;
+
+		running = false;
+	}
+
+	@Override
+	public final void start(LeaderContender contender) throws Exception {
+		Preconditions.checkNotNull(contender, "Contender must not be null.");
+		Preconditions.checkState(leaderContender == null, "Contender was already set.");
+
+		logger.info("Starting LeaderElectionService {}.", this);
+
+		synchronized (lock) {
+			leaderContender = contender;
+			running = true;
+			internalStart(contender);
+		}
+	}
+
+	@Override
+	public final void stop() throws Exception {
+		logger.info("Stopping LeaderElectionService {}.", this);
+
+		synchronized (lock) {
+			if (!running) {
+				return;
+			}
+			running = false;
+			clearConfirmedLeaderInformation();
+			internalStop();
+		}
+	}
+
+	@Override
+	public void confirmLeadership(UUID leaderSessionID, String leaderAddress) {
+		if (logger.isDebugEnabled()) {
+			logger.debug(
+				"Confirm leader session ID {} for leader {}.",
+				leaderSessionID,
+				leaderAddress);
+		}
+
+		Preconditions.checkNotNull(leaderSessionID);
+
+		if (hasLeadership()) {
+			// check if this is an old confirmation call
+			synchronized (lock) {
+				if (running) {
+					if (leaderSessionID.equals(this.issuedLeaderSessionID)) {
+						confirmLeaderInformation(leaderSessionID, leaderAddress);
+						writeLeaderInformation();
+					}
+				} else {
+					logger.debug("Ignoring the leader session Id {} confirmation, since the " +
+						"LeaderElectionService has already been stopped.", leaderSessionID);
+				}
+			}
+		} else {
+			logger.warn("The leader session ID {} was confirmed even though the " +
+				"corresponding JobManager was not elected as the leader.", leaderSessionID);
+		}
+	}
+
+	@Override
+	public final boolean hasLeadership(@Nonnull UUID leaderSessionId) {
+		return hasLeadership() && leaderSessionId.equals(issuedLeaderSessionID);
+	}
+
+	/**
+	 * Returns the current leader session ID or null, if the contender is not the leader.
+	 *
+	 * @return The last leader session ID or null, if the contender is not the leader
+	 */
+	public UUID getLeaderSessionID() {
+		return confirmedLeaderSessionID;
+	}
+
+	/**
+	 * Start the necessary services for specific {@link LeaderElectionService} implementations. For example, LeaderLatch
+	 * and NodeCache in Zookeeper, KubernetesLeaderElector and ConfigMap watcher in Kubernetes. When the leader election
+	 * finished, the contender will be used to grant/revoke leadership.
+	 *
+	 * @param contender LeaderContender to take part in the leader election process.
+	 * @throws Exception Throw exception when start the services.
+	 */
+	@GuardedBy("lock")
+	protected abstract void internalStart(LeaderContender contender) throws Exception;
+
+	/**
+	 * Stop the services used for leader election.
+	 */
+	@GuardedBy("lock")
+	protected abstract void internalStop() throws Exception;
+
+	/**
+	 * Write the current leader information to external persistent storage(e.g. Zookeeper, Kubernetes ConfigMap). This
+	 * is a blocking IO operation.
+	 */
+	protected abstract void writeLeaderInformation();
+
+	/**
+	 * Check whether the current internal leader elector(e.g. Zookeeper LeaderLatch, KubernetesLeaderElector) still have
+	 * the leadership.
+	 *
+	 * @return Whether the current internal leader elector has the leadership.
+	 */
+	protected abstract boolean hasLeadership();
+
+	/**
+	 * Called by subclass when the leadership is granted.
+	 */
+	@GuardedBy("lock")
+	protected final void onGrantLeadership() {
+		synchronized (lock) {
+			if (running) {
+				issuedLeaderSessionID = UUID.randomUUID();
+				clearConfirmedLeaderInformation();
+
+				if (logger.isDebugEnabled()) {
+					logger.debug(
+						"Grant leadership to contender {} with session ID {}.",
+						leaderContender.getDescription(),
+						issuedLeaderSessionID);
+				}
+
+				leaderContender.grantLeadership(issuedLeaderSessionID);
+			} else {
+				logger.debug("Ignoring the grant leadership notification since the service has " +
+					"already been stopped.");
+			}
+		}
+	}
+
+	/**
+	 * Called by subclass when the leadership is revoked.
+	 */
+	@GuardedBy("lock")
+	protected final void onRevokeLeadership() {
+		synchronized (lock) {
+			if (running) {
+				logger.debug(
+					"Revoke leadership of {} ({}@{}).",
+					leaderContender.getDescription(),
+					confirmedLeaderSessionID,
+					confirmedLeaderAddress);
+
+				issuedLeaderSessionID = null;
+				clearConfirmedLeaderInformation();
+
+				leaderContender.revokeLeadership();
+			} else {
+				logger.debug("Ignoring the revoke leadership notification since the service " +
+					"has already been stopped.");
+			}
+		}
+	}
+
+	private void confirmLeaderInformation(UUID leaderSessionID, String leaderAddress) {
+		confirmedLeaderSessionID = leaderSessionID;
+		confirmedLeaderAddress = leaderAddress;
+	}
+
+	private void clearConfirmedLeaderInformation() {
+		confirmedLeaderSessionID = null;
+		confirmedLeaderAddress = null;
+	}

Review comment:
       I'd suggest to annotate these methods with `@GuardedBy("lock")`. Otherwise the analyzer will complain that the fields are accessed w/o having the lock context.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/KubernetesLeaderElectionConfiguration.java
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesHighAvailabilityOptions;
+
+import java.time.Duration;
+
+/**
+ * Configuration specific to {@link org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector}.
+ */
+public class KubernetesLeaderElectionConfiguration {
+
+	private final String clusterId;
+	private final String configMapName;
+	private final Duration leaseDuration;
+	private final Duration renewDeadline;
+	private final Duration retryPeriod;
+
+	private KubernetesLeaderElectionConfiguration(
+			String clusterId,
+			String configMapName,
+			Duration leaseDuration,
+			Duration renewDeadline,
+			Duration retryPeriod) {
+		this.clusterId = clusterId;
+		this.configMapName = configMapName;
+		this.leaseDuration = leaseDuration;
+		this.renewDeadline = renewDeadline;
+		this.retryPeriod = retryPeriod;
+	}
+
+	public String getClusterId() {
+		return clusterId;
+	}
+
+	public String getConfigMapName() {
+		return configMapName;
+	}
+
+	public Duration getLeaseDuration() {
+		return leaseDuration;
+	}
+
+	public Duration getRenewDeadline() {
+		return renewDeadline;
+	}
+
+	public Duration getRetryPeriod() {
+		return retryPeriod;
+	}
+
+	public static KubernetesLeaderElectionConfiguration fromConfiguration(String configMapName, Configuration config) {

Review comment:
       nit: I would invert the order of the parameters because the method is called `fromConfiguration`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,71 @@ public KubernetesWatch watchPodsAndDoCallback(
 				.watch(new KubernetesPodsWatcher(podCallbackHandler)));
 	}
 
+	@Override
+	public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
+		final String configMapName = configMap.getName();
+		return CompletableFuture.runAsync(
+			() -> this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+			kubeClientExecutorService)
+			.whenComplete((ignored, throwable) -> {
+				if (throwable != null) {
+					throw new FlinkRuntimeException("Failed to create ConfigMap " + configMapName, throwable);
+				}
+			});
+	}
+
+	@Override
+	public Optional<KubernetesConfigMap> getConfigMap(String name) {
+		final ConfigMap configMap = this.internalClient.configMaps().inNamespace(namespace).withName(name).get();
+		return configMap == null ? Optional.empty() : Optional.of(new KubernetesConfigMap(configMap));
+	}
+
+	@Override
+	public CompletableFuture<Boolean> checkAndUpdateConfigMap(
+			String configMapName,
+			FunctionWithException<KubernetesConfigMap, Optional<KubernetesConfigMap>, ?> function) {
+		return FutureUtils.retry(
+			() -> CompletableFuture.supplyAsync(
+				() -> getConfigMap(configMapName)
+					.map(FunctionUtils.uncheckedFunction(configMap -> {
+						final boolean updated = function.apply(configMap).map(
+							updatedConfigMap -> {
+								this.internalClient.configMaps()
+									.inNamespace(namespace)
+									.createOrReplace(updatedConfigMap.getInternalResource());
+								return true;
+							}).orElse(false);
+						if (!updated) {
+							LOG.warn("Trying to update ConfigMap {} to {} without checking pass, ignoring.",

Review comment:
       The log message could be `"Failed to update ConfigMap {} with data {} because of concurrent modifications. Trying again."`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,71 @@ public KubernetesWatch watchPodsAndDoCallback(
 				.watch(new KubernetesPodsWatcher(podCallbackHandler)));
 	}
 
+	@Override
+	public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
+		final String configMapName = configMap.getName();
+		return CompletableFuture.runAsync(
+			() -> this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+			kubeClientExecutorService)
+			.whenComplete((ignored, throwable) -> {
+				if (throwable != null) {
+					throw new FlinkRuntimeException("Failed to create ConfigMap " + configMapName, throwable);
+				}
+			});
+	}
+
+	@Override
+	public Optional<KubernetesConfigMap> getConfigMap(String name) {
+		final ConfigMap configMap = this.internalClient.configMaps().inNamespace(namespace).withName(name).get();
+		return configMap == null ? Optional.empty() : Optional.of(new KubernetesConfigMap(configMap));
+	}
+
+	@Override
+	public CompletableFuture<Boolean> checkAndUpdateConfigMap(
+			String configMapName,
+			FunctionWithException<KubernetesConfigMap, Optional<KubernetesConfigMap>, ?> function) {
+		return FutureUtils.retry(
+			() -> CompletableFuture.supplyAsync(
+				() -> getConfigMap(configMapName)
+					.map(FunctionUtils.uncheckedFunction(configMap -> {
+						final boolean updated = function.apply(configMap).map(
+							updatedConfigMap -> {
+								this.internalClient.configMaps()
+									.inNamespace(namespace)
+									.createOrReplace(updatedConfigMap.getInternalResource());
+								return true;
+							}).orElse(false);
+						if (!updated) {
+							LOG.warn("Trying to update ConfigMap {} to {} without checking pass, ignoring.",

Review comment:
       If concurrent writes can happen, then we should not log on `warn` level. Debug log level should be fine.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {

Review comment:
       Similar to the AbstractLeaderElectionService, I am wondering whether composition wouldn't make more sense than inheritance here.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	/** The executor to run external IO operations on. */
+	protected final Executor executor;

Review comment:
       let's call it `ioExecutor`.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	/** The executor to run external IO operations on. */
+	protected final Executor executor;
+
+	/** The runtime configuration. */
+	protected final Configuration configuration;
+
+	/** Store for arbitrary blobs. */
+	private final BlobStoreService blobStoreService;
+
+	public AbstractHaServices(Executor executor, Configuration config, BlobStoreService blobStoreService) {
+
+		this.executor = checkNotNull(executor);
+		this.configuration = checkNotNull(config);
+		this.blobStoreService = blobStoreService;

Review comment:
       If the other parameters are checked, then we should also null check `blobStoreService`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,71 @@ public KubernetesWatch watchPodsAndDoCallback(
 				.watch(new KubernetesPodsWatcher(podCallbackHandler)));
 	}
 
+	@Override
+	public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
+		final String configMapName = configMap.getName();
+		return CompletableFuture.runAsync(
+			() -> this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+			kubeClientExecutorService)
+			.whenComplete((ignored, throwable) -> {
+				if (throwable != null) {
+					throw new FlinkRuntimeException("Failed to create ConfigMap " + configMapName, throwable);
+				}
+			});

Review comment:
       I think this pattern won't work. I'd suggest to use `exceptionally`.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	/** The executor to run external IO operations on. */
+	protected final Executor executor;
+
+	/** The runtime configuration. */
+	protected final Configuration configuration;
+
+	/** Store for arbitrary blobs. */
+	private final BlobStoreService blobStoreService;
+
+	public AbstractHaServices(Executor executor, Configuration config, BlobStoreService blobStoreService) {
+
+		this.executor = checkNotNull(executor);
+		this.configuration = checkNotNull(config);
+		this.blobStoreService = blobStoreService;
+	}
+
+	@Override
+	public LeaderRetrievalService getResourceManagerLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForResourceManager());
+	}
+
+	@Override
+	public LeaderRetrievalService getDispatcherLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForDispatcher());
+	}
+
+	@Override
+	public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID) {
+		return createLeaderRetrievalService(getLeaderNameForJobManager(jobID));
+	}
+
+	@Override
+	public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+		return getJobManagerLeaderRetriever(jobID);
+	}
+
+	@Override
+	public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForRestServer());
+	}
+
+	@Override
+	public LeaderElectionService getResourceManagerLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForResourceManager());
+	}
+
+	@Override
+	public LeaderElectionService getDispatcherLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForDispatcher());
+	}
+
+	@Override
+	public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
+		return createLeaderElectionService(getLeaderNameForJobManager(jobID));
+	}
+
+	@Override
+	public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForRestServer());
+	}
+
+	@Override
+	public BlobStore createBlobStore() {
+		return blobStoreService;
+	}
+
+	@Override
+	public void close() throws Exception {
+		Throwable exception = null;
+
+		try {
+			blobStoreService.close();
+		} catch (Throwable t) {
+			exception = t;
+		}
+
+		internalClose();
+
+		if (exception != null) {
+			ExceptionUtils.rethrowException(exception, "Could not properly close the " + getClass().getSimpleName());
+		}
+	}
+
+	@Override
+	public void closeAndCleanupAllData() throws Exception {
+		logger.info("Close and clean up all data for {}.", getClass().getSimpleName());
+
+		Throwable exception = null;
+
+		try {
+			blobStoreService.closeAndCleanupAllData();
+		} catch (Throwable t) {
+			exception = t;
+		}
+
+		try {
+			internalCleanup();
+		} catch (Throwable t) {
+			exception = ExceptionUtils.firstOrSuppressed(t, exception);
+		}
+
+		internalClose();
+
+		if (exception != null) {
+			ExceptionUtils.rethrowException(
+				exception, "Could not properly close and clean up all data of high availability service.");
+		}
+		logger.info("Finished cleaning up the high availability data.");
+	}
+
+	/**
+	 * Create leader election service with specified leaderName.
+	 * @param leaderName ConfigMap name in Kubernetes or child node path in Zookeeper.
+	 *
+	 * @return Return LeaderElectionService using Zookeeper or Kubernetes.
+	 */
+	protected abstract LeaderElectionService createLeaderElectionService(String leaderName);
+
+	/**
+	 * Create leader retrieval service with specified leaderName.
+	 * @param leaderName ConfigMap name in Kubernetes or child node path in Zookeeper.
+	 *
+	 * @return Return LeaderRetrievalService using Zookeeper or Kubernetes.
+	 */
+	protected abstract LeaderRetrievalService createLeaderRetrievalService(String leaderName);
+
+	/**
+	 * Closes components which don't distinguish between close and closeAndCleanupAllData.
+	 */
+	protected abstract void internalClose() throws Exception;

Review comment:
       Would be good to state what happens if this method throws an `Exception`.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/AbstractHaServices.java
##########
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobStore;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Abstract high availability services based on distributed system(e.g. Zookeeper, Kubernetes).
+ */
+public abstract class AbstractHaServices implements HighAvailabilityServices {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	/** The executor to run external IO operations on. */
+	protected final Executor executor;
+
+	/** The runtime configuration. */
+	protected final Configuration configuration;
+
+	/** Store for arbitrary blobs. */
+	private final BlobStoreService blobStoreService;
+
+	public AbstractHaServices(Executor executor, Configuration config, BlobStoreService blobStoreService) {
+
+		this.executor = checkNotNull(executor);
+		this.configuration = checkNotNull(config);
+		this.blobStoreService = blobStoreService;
+	}
+
+	@Override
+	public LeaderRetrievalService getResourceManagerLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForResourceManager());
+	}
+
+	@Override
+	public LeaderRetrievalService getDispatcherLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForDispatcher());
+	}
+
+	@Override
+	public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID) {
+		return createLeaderRetrievalService(getLeaderNameForJobManager(jobID));
+	}
+
+	@Override
+	public LeaderRetrievalService getJobManagerLeaderRetriever(JobID jobID, String defaultJobManagerAddress) {
+		return getJobManagerLeaderRetriever(jobID);
+	}
+
+	@Override
+	public LeaderRetrievalService getClusterRestEndpointLeaderRetriever() {
+		return createLeaderRetrievalService(getLeaderNameForRestServer());
+	}
+
+	@Override
+	public LeaderElectionService getResourceManagerLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForResourceManager());
+	}
+
+	@Override
+	public LeaderElectionService getDispatcherLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForDispatcher());
+	}
+
+	@Override
+	public LeaderElectionService getJobManagerLeaderElectionService(JobID jobID) {
+		return createLeaderElectionService(getLeaderNameForJobManager(jobID));
+	}
+
+	@Override
+	public LeaderElectionService getClusterRestEndpointLeaderElectionService() {
+		return createLeaderElectionService(getLeaderNameForRestServer());
+	}
+
+	@Override
+	public BlobStore createBlobStore() {
+		return blobStoreService;
+	}
+
+	@Override
+	public void close() throws Exception {
+		Throwable exception = null;
+
+		try {
+			blobStoreService.close();
+		} catch (Throwable t) {
+			exception = t;
+		}
+
+		internalClose();
+
+		if (exception != null) {
+			ExceptionUtils.rethrowException(exception, "Could not properly close the " + getClass().getSimpleName());
+		}
+	}
+
+	@Override
+	public void closeAndCleanupAllData() throws Exception {
+		logger.info("Close and clean up all data for {}.", getClass().getSimpleName());
+
+		Throwable exception = null;
+
+		try {
+			blobStoreService.closeAndCleanupAllData();
+		} catch (Throwable t) {
+			exception = t;
+		}
+
+		try {
+			internalCleanup();
+		} catch (Throwable t) {
+			exception = ExceptionUtils.firstOrSuppressed(t, exception);
+		}
+
+		internalClose();
+
+		if (exception != null) {
+			ExceptionUtils.rethrowException(
+				exception, "Could not properly close and clean up all data of high availability service.");
+		}
+		logger.info("Finished cleaning up the high availability data.");
+	}
+
+	/**
+	 * Create leader election service with specified leaderName.
+	 * @param leaderName ConfigMap name in Kubernetes or child node path in Zookeeper.
+	 *
+	 * @return Return LeaderElectionService using Zookeeper or Kubernetes.
+	 */
+	protected abstract LeaderElectionService createLeaderElectionService(String leaderName);
+
+	/**
+	 * Create leader retrieval service with specified leaderName.
+	 * @param leaderName ConfigMap name in Kubernetes or child node path in Zookeeper.
+	 *
+	 * @return Return LeaderRetrievalService using Zookeeper or Kubernetes.
+	 */
+	protected abstract LeaderRetrievalService createLeaderRetrievalService(String leaderName);
+
+	/**
+	 * Closes components which don't distinguish between close and closeAndCleanupAllData.
+	 */
+	protected abstract void internalClose() throws Exception;
+
+	/**
+	 * Clean up the meta data in the distributed system(e.g. Zookeeper, Kubernetes ConfigMap).
+	 */
+	protected abstract void internalCleanup() throws Exception;

Review comment:
       Same here.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesConfigMapWatcher.java
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.resources;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Watcher for {@link ConfigMap}s in Kubernetes.

Review comment:
       `{@link ConfigMap ConfigMaps}` solves the problem of a plural here.

##########
File path: docs/_includes/generated/expert_high_availability_k8s_section.html
##########
@@ -0,0 +1,30 @@
+<table class="table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.lease-duration</h5></td>
+            <td style="word-wrap: break-word;">15 s</td>
+            <td>Duration</td>
+            <td>Define the lease duration for the Kubernetes leader election in ms. The leader will continuously renew its lease time to indicate its existence. And the followers will do a lease checking against the current time. "renewTime + leaseDuration &gt; now" means the leader is alive.</td>

Review comment:
       `in s` instead of `in ms`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,71 @@ public KubernetesWatch watchPodsAndDoCallback(
 				.watch(new KubernetesPodsWatcher(podCallbackHandler)));
 	}
 
+	@Override
+	public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
+		final String configMapName = configMap.getName();
+		return CompletableFuture.runAsync(
+			() -> this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+			kubeClientExecutorService)
+			.whenComplete((ignored, throwable) -> {
+				if (throwable != null) {
+					throw new FlinkRuntimeException("Failed to create ConfigMap " + configMapName, throwable);
+				}
+			});
+	}
+
+	@Override
+	public Optional<KubernetesConfigMap> getConfigMap(String name) {
+		final ConfigMap configMap = this.internalClient.configMaps().inNamespace(namespace).withName(name).get();
+		return configMap == null ? Optional.empty() : Optional.of(new KubernetesConfigMap(configMap));
+	}
+
+	@Override
+	public CompletableFuture<Boolean> checkAndUpdateConfigMap(
+			String configMapName,
+			FunctionWithException<KubernetesConfigMap, Optional<KubernetesConfigMap>, ?> function) {
+		return FutureUtils.retry(
+			() -> CompletableFuture.supplyAsync(
+				() -> getConfigMap(configMapName)
+					.map(FunctionUtils.uncheckedFunction(configMap -> {
+						final boolean updated = function.apply(configMap).map(
+							updatedConfigMap -> {
+								this.internalClient.configMaps()
+									.inNamespace(namespace)
+									.createOrReplace(updatedConfigMap.getInternalResource());
+								return true;
+							}).orElse(false);
+						if (!updated) {
+							LOG.warn("Trying to update ConfigMap {} to {} without checking pass, ignoring.",
+								configMap.getName(), configMap.getData());
+						}
+						return updated;
+					}))
+					.orElseThrow(
+						() -> new FlinkRuntimeException("ConfigMap " + configMapName + " not exists.")),

Review comment:
       Does it make sense to retry this operation if the config map does not exist? Maybe we should stop the operation right after realizing the non-existence of the config map.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/AbstractLeaderElectionService.java
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.leaderelection;
+
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.UUID;
+
+/**
+ * Abstract class for leader election service based on distributed coordination system(e.g. Zookeeper, Kubernetes, etc.).
+ */
+public abstract class AbstractLeaderElectionService implements LeaderElectionService {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	protected final Object lock = new Object();
+
+	/** The leader contender which applies for leadership. */
+	@GuardedBy("lock")
+	protected volatile LeaderContender leaderContender;
+
+	@GuardedBy("lock")
+	private volatile UUID issuedLeaderSessionID;
+
+	@GuardedBy("lock")
+	protected volatile UUID confirmedLeaderSessionID;
+
+	@GuardedBy("lock")
+	protected volatile String confirmedLeaderAddress;
+
+	@GuardedBy("lock")
+	protected volatile boolean running;
+
+	protected AbstractLeaderElectionService() {
+		leaderContender = null;
+
+		issuedLeaderSessionID = null;
+		confirmedLeaderSessionID = null;
+		confirmedLeaderAddress = null;
+
+		running = false;
+	}
+
+	@Override
+	public final void start(LeaderContender contender) throws Exception {
+		Preconditions.checkNotNull(contender, "Contender must not be null.");
+		Preconditions.checkState(leaderContender == null, "Contender was already set.");
+
+		logger.info("Starting LeaderElectionService {}.", this);
+
+		synchronized (lock) {
+			leaderContender = contender;
+			running = true;
+			internalStart(contender);
+		}
+	}
+
+	@Override
+	public final void stop() throws Exception {
+		logger.info("Stopping LeaderElectionService {}.", this);
+
+		synchronized (lock) {
+			if (!running) {
+				return;
+			}
+			running = false;
+			clearConfirmedLeaderInformation();
+			internalStop();
+		}
+	}
+
+	@Override
+	public void confirmLeadership(UUID leaderSessionID, String leaderAddress) {
+		if (logger.isDebugEnabled()) {
+			logger.debug(
+				"Confirm leader session ID {} for leader {}.",
+				leaderSessionID,
+				leaderAddress);
+		}
+
+		Preconditions.checkNotNull(leaderSessionID);
+
+		if (hasLeadership()) {
+			// check if this is an old confirmation call
+			synchronized (lock) {
+				if (running) {
+					if (leaderSessionID.equals(this.issuedLeaderSessionID)) {
+						confirmLeaderInformation(leaderSessionID, leaderAddress);
+						writeLeaderInformation();
+					}
+				} else {
+					logger.debug("Ignoring the leader session Id {} confirmation, since the " +
+						"LeaderElectionService has already been stopped.", leaderSessionID);
+				}
+			}
+		} else {
+			logger.warn("The leader session ID {} was confirmed even though the " +
+				"corresponding JobManager was not elected as the leader.", leaderSessionID);
+		}
+	}
+
+	@Override
+	public final boolean hasLeadership(@Nonnull UUID leaderSessionId) {
+		return hasLeadership() && leaderSessionId.equals(issuedLeaderSessionID);

Review comment:
       `issuedLeaderSessionID` is accessed outside of `lock` scope.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.
+	 *
+	 * @return Return the ConfigMap, or empty if the ConfigMap does not exist.
+	 */
+	Optional<KubernetesConfigMap> getConfigMap(String name);
+
+	/**
+	 * Update an existing ConfigMap with the data. Benefit from <a href=https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions>
+	 * resource version</a> and combined with {@link #getConfigMap(String)}, we could perform a get-check-and-update
+	 * transactional operation. Since concurrent modification could happen on a same ConfigMap,
+	 * the update operation may fail. We need to retry internally. The max retry attempts could be
+	 * configured via {@link org.apache.flink.kubernetes.configuration.KubernetesConfigOptions#KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES}.
+	 *
+	 * @param configMapName ConfigMap to be replaced with.
+	 * @param function      Function to be applied to the obtained ConfigMap and get a new updated one. If the returned

Review comment:
       Maybe call this parameter `updateFunction`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,71 @@ public KubernetesWatch watchPodsAndDoCallback(
 				.watch(new KubernetesPodsWatcher(podCallbackHandler)));
 	}
 
+	@Override
+	public CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap) {
+		final String configMapName = configMap.getName();
+		return CompletableFuture.runAsync(
+			() -> this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+			kubeClientExecutorService)
+			.whenComplete((ignored, throwable) -> {
+				if (throwable != null) {
+					throw new FlinkRuntimeException("Failed to create ConfigMap " + configMapName, throwable);
+				}
+			});
+	}
+
+	@Override
+	public Optional<KubernetesConfigMap> getConfigMap(String name) {
+		final ConfigMap configMap = this.internalClient.configMaps().inNamespace(namespace).withName(name).get();
+		return configMap == null ? Optional.empty() : Optional.of(new KubernetesConfigMap(configMap));
+	}
+
+	@Override
+	public CompletableFuture<Boolean> checkAndUpdateConfigMap(
+			String configMapName,
+			FunctionWithException<KubernetesConfigMap, Optional<KubernetesConfigMap>, ?> function) {
+		return FutureUtils.retry(
+			() -> CompletableFuture.supplyAsync(
+				() -> getConfigMap(configMapName)
+					.map(FunctionUtils.uncheckedFunction(configMap -> {
+						final boolean updated = function.apply(configMap).map(
+							updatedConfigMap -> {
+								this.internalClient.configMaps()
+									.inNamespace(namespace)
+									.createOrReplace(updatedConfigMap.getInternalResource());
+								return true;
+							}).orElse(false);
+						if (!updated) {
+							LOG.warn("Trying to update ConfigMap {} to {} without checking pass, ignoring.",
+								configMap.getName(), configMap.getData());
+						}
+						return updated;
+					}))
+					.orElseThrow(
+						() -> new FlinkRuntimeException("ConfigMap " + configMapName + " not exists.")),

Review comment:
       `ConfigMap ... does not exist.`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderElectionService{configMapName='" + configMapName + "'}";
+	}

Review comment:
       For what do we need the `toString` method?

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -249,4 +265,69 @@ public void testStopAndCleanupCluster() throws Exception {
 		this.flinkKubeClient.stopAndCleanupCluster(CLUSTER_ID);
 		assertTrue(this.kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems().isEmpty());
 	}
+
+	@Test
+	public void testCreateConfigMap() throws Exception {
+		final KubernetesConfigMap configMap = buildTestingConfigMap();
+		this.flinkKubeClient.createConfigMap(configMap).get();
+		final Optional<KubernetesConfigMap> currentOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(currentOpt.isPresent(), is(true));
+		assertThat(currentOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_VALUE));
+	}
+
+	@Test
+	public void testDeleteConfigMapByLabels() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMapsByLabels(TESTING_LABELS);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testDeleteConfigMapByName() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testCheckAndUpdateConfigMap() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap());
+
+		// Checker pass
+		this.flinkKubeClient.checkAndUpdateConfigMap(
+			TESTING_CONFIG_MAP_NAME,
+			c -> {
+				c.getData().put(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_NEW_VALUE);
+				return Optional.of(c);
+			}).get();
+
+		final Optional<KubernetesConfigMap> configMapOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(configMapOpt.isPresent(), is(true));
+		assertThat(configMapOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_NEW_VALUE));
+	}
+
+	@Test
+	public void testCheckAndUpdateConfigMapWithEmptyResult() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap());
+
+		// Checker not pass and return empty result
+		this.flinkKubeClient.checkAndUpdateConfigMap(TESTING_CONFIG_MAP_NAME, c -> Optional.empty()).get();
+
+		final Optional<KubernetesConfigMap> configMapOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(configMapOpt.isPresent(), is(true));
+		assertThat(configMapOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_VALUE));
+	}
+
+	private KubernetesConfigMap buildTestingConfigMap() {
+		final Map<String, String> data = new HashMap<>();
+		data.put(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_VALUE);
+		return new KubernetesConfigMap(new ConfigMapBuilder()
+			.withNewMetadata()
+			.withName(TESTING_CONFIG_MAP_NAME)
+			.withLabels(TESTING_LABELS)
+			.endMetadata()
+			.withData(data).build());
+	}

Review comment:
       Maybe add a test that concurrent writes will be resolved and don't overwrite each other.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/AbstractLeaderElectionService.java
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.leaderelection;
+
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.UUID;
+
+/**
+ * Abstract class for leader election service based on distributed coordination system(e.g. Zookeeper, Kubernetes, etc.).
+ */
+public abstract class AbstractLeaderElectionService implements LeaderElectionService {
+
+	protected final Logger logger = LoggerFactory.getLogger(getClass());
+
+	protected final Object lock = new Object();

Review comment:
       Ensuring the correct usage of synchronization patterns across sub classes is usually very hard and brittle because you never know which other sub class will be added later. The same applies a bit to exposing more or less all the fields of this class to the sub classes.
   
   Have you considered using composition instead of inheritance to achieve the goal of sharing the common logic? I think it should be feasible to have a `LeaderElectionService` which takes system dependent connector (Zk, K8s) which is responsible for the interaction with the external system. Similar to what Xintong did with the `ResourceManager` and the `ResourceManagerDrivers`.
   
   The benefit of using composition would be that we don't expose internal details of this class. Ideally, we would keep the synchronization logic all contained in this class.

##########
File path: docs/_includes/generated/expert_high_availability_k8s_section.html
##########
@@ -0,0 +1,30 @@
+<table class="table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.lease-duration</h5></td>
+            <td style="word-wrap: break-word;">15 s</td>
+            <td>Duration</td>
+            <td>Define the lease duration for the Kubernetes leader election in ms. The leader will continuously renew its lease time to indicate its existence. And the followers will do a lease checking against the current time. "renewTime + leaseDuration &gt; now" means the leader is alive.</td>
+        </tr>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.renew-deadline</h5></td>
+            <td style="word-wrap: break-word;">15 s</td>
+            <td>Duration</td>
+            <td>Defines the deadline when the leader tries to renew the lease in ms. The leader will give up its leadership if it cannot successfully renew the lease in the given time.</td>
+        </tr>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.retry-period</h5></td>
+            <td style="word-wrap: break-word;">5 s</td>
+            <td>Duration</td>
+            <td>Defines the pause between consecutive retries in ms. All the contenders, including the current leader and all other followers, periodically try to acquire/renew the leadership if possible at this interval.</td>

Review comment:
       `in s` instead of `in ms`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.
+	 *
+	 * @return Return the ConfigMap, or empty if the ConfigMap does not exist.
+	 */
+	Optional<KubernetesConfigMap> getConfigMap(String name);
+
+	/**
+	 * Update an existing ConfigMap with the data. Benefit from <a href=https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions>
+	 * resource version</a> and combined with {@link #getConfigMap(String)}, we could perform a get-check-and-update
+	 * transactional operation. Since concurrent modification could happen on a same ConfigMap,
+	 * the update operation may fail. We need to retry internally. The max retry attempts could be
+	 * configured via {@link org.apache.flink.kubernetes.configuration.KubernetesConfigOptions#KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES}.
+	 *
+	 * @param configMapName ConfigMap to be replaced with.

Review comment:
       `@param configMapName configMapName specifies the name of the ConfigMap which shall be updated`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesHaServices.java
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.blob.BlobStoreService;
+import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
+import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
+import org.apache.flink.runtime.highavailability.AbstractHaServices;
+import org.apache.flink.runtime.highavailability.RunningJobsRegistry;
+import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneRunningJobsRegistry;
+import org.apache.flink.runtime.jobmanager.JobGraphStore;
+import org.apache.flink.runtime.jobmanager.StandaloneJobGraphStore;
+import org.apache.flink.runtime.leaderelection.LeaderElectionService;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.NAME_SEPARATOR;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * High availability service for Kubernetes.
+ */
+public class KubernetesHaServices extends AbstractHaServices {
+
+	private final String clusterId;
+
+	/** Kubernetes client. */
+	private final FlinkKubeClient kubeClient;
+
+	private static final String RESOURCE_MANAGER_NAME = "resourcemanager";
+
+	private static final String DISPATCHER_NAME = "dispatcher";
+
+	private static final String JOB_MANAGER_NAME = "jobmanager";
+
+	private static final String REST_SERVER_NAME = "restserver";
+
+	private static final String LEADER_SUFFIX = "leader";
+
+	KubernetesHaServices(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			Configuration config,
+			BlobStoreService blobStoreService) {
+
+		super(executor, config, blobStoreService);
+		this.kubeClient = checkNotNull(kubeClient);
+		this.clusterId = checkNotNull(config.get(KubernetesConfigOptions.CLUSTER_ID));
+	}
+
+	@Override
+	public CheckpointRecoveryFactory getCheckpointRecoveryFactory() {
+		return new StandaloneCheckpointRecoveryFactory();
+	}
+
+	@Override
+	public JobGraphStore getJobGraphStore() {
+		return new StandaloneJobGraphStore();
+	}
+
+	@Override
+	public RunningJobsRegistry getRunningJobsRegistry() {
+		return new StandaloneRunningJobsRegistry();
+	}
+
+	@Override
+	public LeaderElectionService createLeaderElectionService(String leaderName) {
+		return new KubernetesLeaderElectionService(
+			kubeClient,
+			executor,
+			KubernetesLeaderElectionConfiguration.fromConfiguration(leaderName, configuration));
+	}
+
+	@Override
+	public LeaderRetrievalService createLeaderRetrievalService(String leaderName) {
+		return new KubernetesLeaderRetrievalService(kubeClient, leaderName);
+	}
+
+	@Override
+	public void internalClose() throws Exception {
+		kubeClient.close();

Review comment:
       Due to this call, `KubeClientFactory.getFlinkKubeClient` must not return a singleton. If the `KubernetesHaServices` own the `kubeClient`, then we need to create separate clients.

##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/AbstractLeaderElectionService.java
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.leaderelection;
+
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.UUID;
+
+/**
+ * Abstract class for leader election service based on distributed coordination system(e.g. Zookeeper, Kubernetes, etc.).
+ */

Review comment:
       I think this JavaDoc should state the contracts of this class and what a user needs to pay attention to when writing a sub class, in particular since we expose the `lock` and most of its fields.

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -249,4 +265,69 @@ public void testStopAndCleanupCluster() throws Exception {
 		this.flinkKubeClient.stopAndCleanupCluster(CLUSTER_ID);
 		assertTrue(this.kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems().isEmpty());
 	}
+
+	@Test
+	public void testCreateConfigMap() throws Exception {
+		final KubernetesConfigMap configMap = buildTestingConfigMap();
+		this.flinkKubeClient.createConfigMap(configMap).get();
+		final Optional<KubernetesConfigMap> currentOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(currentOpt.isPresent(), is(true));
+		assertThat(currentOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_VALUE));
+	}
+
+	@Test
+	public void testDeleteConfigMapByLabels() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMapsByLabels(TESTING_LABELS);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testDeleteConfigMapByName() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testCheckAndUpdateConfigMap() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap());
+
+		// Checker pass
+		this.flinkKubeClient.checkAndUpdateConfigMap(
+			TESTING_CONFIG_MAP_NAME,
+			c -> {
+				c.getData().put(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_NEW_VALUE);
+				return Optional.of(c);
+			}).get();
+
+		final Optional<KubernetesConfigMap> configMapOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(configMapOpt.isPresent(), is(true));
+		assertThat(configMapOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_NEW_VALUE));
+	}
+
+	@Test
+	public void testCheckAndUpdateConfigMapWithEmptyResult() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap());
+
+		// Checker not pass and return empty result
+		this.flinkKubeClient.checkAndUpdateConfigMap(TESTING_CONFIG_MAP_NAME, c -> Optional.empty()).get();

Review comment:
       Maybe assert that this returns `false`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;

Review comment:
       `@Nullable`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.
+	 *
+	 * @return Return the ConfigMap, or empty if the ConfigMap does not exist.
+	 */
+	Optional<KubernetesConfigMap> getConfigMap(String name);
+
+	/**
+	 * Update an existing ConfigMap with the data. Benefit from <a href=https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions>
+	 * resource version</a> and combined with {@link #getConfigMap(String)}, we could perform a get-check-and-update
+	 * transactional operation. Since concurrent modification could happen on a same ConfigMap,
+	 * the update operation may fail. We need to retry internally. The max retry attempts could be
+	 * configured via {@link org.apache.flink.kubernetes.configuration.KubernetesConfigOptions#KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES}.

Review comment:
       The configuration is an implementation detail of an implementation of this interface. I think it should not be mentioned in the JavaDocs of this interface. Instead one could say that implementations of this method will retry the operation. The number of retries is implementation specific, though.

##########
File path: docs/_includes/generated/expert_high_availability_k8s_section.html
##########
@@ -0,0 +1,30 @@
+<table class="table table-bordered">
+    <thead>
+        <tr>
+            <th class="text-left" style="width: 20%">Key</th>
+            <th class="text-left" style="width: 15%">Default</th>
+            <th class="text-left" style="width: 10%">Type</th>
+            <th class="text-left" style="width: 55%">Description</th>
+        </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.lease-duration</h5></td>
+            <td style="word-wrap: break-word;">15 s</td>
+            <td>Duration</td>
+            <td>Define the lease duration for the Kubernetes leader election in ms. The leader will continuously renew its lease time to indicate its existence. And the followers will do a lease checking against the current time. "renewTime + leaseDuration &gt; now" means the leader is alive.</td>
+        </tr>
+        <tr>
+            <td><h5>high-availability.kubernetes.leader-election.renew-deadline</h5></td>
+            <td style="word-wrap: break-word;">15 s</td>
+            <td>Duration</td>
+            <td>Defines the deadline when the leader tries to renew the lease in ms. The leader will give up its leadership if it cannot successfully renew the lease in the given time.</td>

Review comment:
       `in s` instead of `in ms`.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.

Review comment:
       nit: this does not really describe what the `configMap` parameter is.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderElectionService{configMapName='" + configMapName + "'}";
+	}
+
+	private class LeaderCallbackHandlerImpl extends KubernetesLeaderElector.LeaderCallbackHandler {
+
+		@Override
+		public void isLeader() {
+			onGrantLeadership();
+		}
+
+		@Override
+		public void notLeader() {
+			// Clear the leader information in ConfigMap
+			try {
+				kubeClient.checkAndUpdateConfigMap(
+					configMapName,
+					configMap -> {
+						// Do not need to check the leader here
+						configMap.getData().remove(LEADER_ADDRESS_KEY);
+						configMap.getData().remove(LEADER_SESSION_ID_KEY);

Review comment:
       Can't it happen that we are removing the leader information of our successor here if this call is late? The next leader might already have written its leader information once this is executed.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);

Review comment:
       What if `leaderElector::run` fails? Should we assert that this won't throw an exception via `FutureUtils.assertNoException`?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.

Review comment:
       Maybe state that the returned future will be completed exceptionally if the ConfigMap already exists.

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -249,4 +265,69 @@ public void testStopAndCleanupCluster() throws Exception {
 		this.flinkKubeClient.stopAndCleanupCluster(CLUSTER_ID);
 		assertTrue(this.kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems().isEmpty());
 	}
+
+	@Test
+	public void testCreateConfigMap() throws Exception {
+		final KubernetesConfigMap configMap = buildTestingConfigMap();
+		this.flinkKubeClient.createConfigMap(configMap).get();
+		final Optional<KubernetesConfigMap> currentOpt = this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(currentOpt.isPresent(), is(true));
+		assertThat(currentOpt.get().getData().get(TESTING_CONFIG_MAP_KEY), is(TESTING_CONFIG_MAP_VALUE));
+	}
+
+	@Test
+	public void testDeleteConfigMapByLabels() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMapsByLabels(TESTING_LABELS);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testDeleteConfigMapByName() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(true));
+		this.flinkKubeClient.deleteConfigMap(TESTING_CONFIG_MAP_NAME);
+		assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME).isPresent(), is(false));
+	}
+
+	@Test
+	public void testCheckAndUpdateConfigMap() throws Exception {
+		this.flinkKubeClient.createConfigMap(buildTestingConfigMap());
+
+		// Checker pass
+		this.flinkKubeClient.checkAndUpdateConfigMap(
+			TESTING_CONFIG_MAP_NAME,
+			c -> {
+				c.getData().put(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_NEW_VALUE);
+				return Optional.of(c);
+			}).get();

Review comment:
       Should we assert that this expression returns `true`?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.
+	 *
+	 * @return Return the ConfigMap, or empty if the ConfigMap does not exist.
+	 */
+	Optional<KubernetesConfigMap> getConfigMap(String name);
+
+	/**
+	 * Update an existing ConfigMap with the data. Benefit from <a href=https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions>
+	 * resource version</a> and combined with {@link #getConfigMap(String)}, we could perform a get-check-and-update
+	 * transactional operation. Since concurrent modification could happen on a same ConfigMap,
+	 * the update operation may fail. We need to retry internally. The max retry attempts could be
+	 * configured via {@link org.apache.flink.kubernetes.configuration.KubernetesConfigOptions#KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES}.
+	 *
+	 * @param configMapName ConfigMap to be replaced with.
+	 * @param function      Function to be applied to the obtained ConfigMap and get a new updated one. If the returned
+	 *                      optional is empty, we will not do the update.

Review comment:
       I'd suggest to not align param descriptions because one changes the parameter name, then it will get quickly out of sync again.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);

Review comment:
       What is the meaning of a non-existent ConfigMap with name `configMapName`? Can this ever happen? Is this an invalid state where we should fail hard?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;

Review comment:
       Why are we setting `leaderContender` here?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {

Review comment:
       Aren't we accessing these fields outside of the guard `lock` here?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.

Review comment:
       `@param name name of the ConfigMap to retrieve`

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.

Review comment:
       Why are the JavaDocs referring to the `JobManagers`? Isn't this service also used for other leader elections?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalService.java
##########
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
+import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The counterpart to the {@link org.apache.flink.kubernetes.highavailability.KubernetesLeaderElectionService}.
+ * This implementation of the {@link LeaderRetrievalService} retrieves the current leader which has
+ * been elected by the {@link org.apache.flink.kubernetes.highavailability.KubernetesLeaderElectionService}.
+ * The leader address as well as the current leader session ID is retrieved from Kubernetes ConfigMap.
+ */
+class KubernetesLeaderRetrievalService implements LeaderRetrievalService {
+
+	private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeaderRetrievalService.class);
+
+	private final Object lock = new Object();
+
+	private final FlinkKubeClient kubeClient;
+
+	private final String configMapName;
+
+	@GuardedBy("lock")
+	private volatile String lastLeaderAddress;
+
+	@GuardedBy("lock")
+	private volatile UUID lastLeaderSessionID;
+
+	@GuardedBy("lock")
+	private volatile LeaderRetrievalListener leaderListener;
+
+	@GuardedBy("lock")
+	private volatile boolean running;
+
+	private KubernetesWatch kubernetesWatch;
+
+	KubernetesLeaderRetrievalService(FlinkKubeClient kubeClient, String configMapName) {
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.configMapName = checkNotNull(configMapName, "ConfigMap name should not be null.");
+
+		this.leaderListener = null;
+		this.lastLeaderAddress = null;
+		this.lastLeaderSessionID = null;
+
+		running = false;
+	}
+
+	@Override
+	public void start(LeaderRetrievalListener listener) {
+		checkNotNull(listener, "Listener must not be null.");
+		Preconditions.checkState(leaderListener == null, "KubernetesLeaderRetrievalService can " +
+			"only be started once.");
+
+		LOG.info("Starting {}.", this);
+
+		synchronized (lock) {
+			running = true;
+			leaderListener = listener;
+			kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+		}
+	}
+
+	@Override
+	public void stop() {
+		LOG.info("Stopping {}.", this);
+
+		synchronized (lock) {
+			if (!running) {
+				return;
+			}
+			running = false;
+			if (kubernetesWatch != null) {
+				kubernetesWatch.close();
+			}
+		}
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderRetrievalService{configMapName='" + configMapName + "'}";
+	}
+
+	private class ConfigMapCallbackHandlerImpl implements FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> {
+
+		@Override
+		public void onAdded(List<KubernetesConfigMap> configMaps) {
+			handleEvent(configMaps);
+		}
+
+		@Override
+		public void onModified(List<KubernetesConfigMap> configMaps) {
+			handleEvent(configMaps);
+		}
+
+		@Override
+		public void onDeleted(List<KubernetesConfigMap> configMaps) {
+			// Nothing to do since a new ConfigMap will be created if it is deleted externally.
+		}
+
+		@Override
+		public void onError(List<KubernetesConfigMap> configMaps) {
+			leaderListener.handleError(new Exception("Error while watching the ConfigMap " + configMapName));
+		}
+
+		@Override
+		public void handleFatalError(Throwable throwable) {
+			leaderListener.handleError(
+				new Exception("Fatal error while watching the ConfigMap " + configMapName, throwable));
+		}
+
+		private void handleEvent(List<KubernetesConfigMap> configMaps) {
+			synchronized (lock) {
+				if (running) {
+					configMaps.forEach(e -> {
+						if (e.getName().equals(configMapName)) {
+							final String leaderAddress = e.getData().get(LEADER_ADDRESS_KEY);
+							final String sessionID = e.getData().get(LEADER_SESSION_ID_KEY);
+							if (leaderAddress != null && sessionID != null) {
+								final UUID leaderSessionID = UUID.fromString(sessionID);
+								if (!(Objects.equals(leaderAddress, lastLeaderAddress) &&
+									Objects.equals(leaderSessionID, lastLeaderSessionID))) {
+									LOG.debug(
+										"New leader information: Leader={}, session ID={}.",
+										leaderAddress,
+										leaderSessionID);
+
+									lastLeaderAddress = leaderAddress;
+									lastLeaderSessionID = leaderSessionID;
+									leaderListener.notifyLeaderAddress(leaderAddress, leaderSessionID);
+								}
+							} else {
+								LOG.debug("Leader information was lost: The listener will be notified accordingly.");
+								leaderListener.notifyLeaderAddress(
+									leaderAddress, sessionID == null ? null : UUID.fromString(sessionID));
+							}

Review comment:
       I guess this logic could be factored out and shared between the Zk and K8s implementations.

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.resources;
+
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfigBuilder;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
+import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+
+/**
+ * Represent {@link KubernetesLeaderElector} in kubernetes. {@link LeaderElector#run()} is a blocking call. It should be
+ *  run in the IO executor, not the main thread. The lifecycle is bound to single leader election. Once the leadership
+ * is revoked, as well as the {@link LeaderCallbackHandler#notLeader()} is called, the {@link LeaderElector#run()} will
+ * finish. To start another round of election, we need to trigger again.
+ */
+public class KubernetesLeaderElector extends LeaderElector<NamespacedKubernetesClient> {
+
+	private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeaderElector.class);
+	protected static final String LOCK_IDENTITY = UUID.randomUUID().toString();

Review comment:
       Won't it be a problem to have only one identity if we have multiple election services for the same type (e.g. multiple JobManagers running in the `MiniCluster`)?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderElectionService{configMapName='" + configMapName + "'}";
+	}
+
+	private class LeaderCallbackHandlerImpl extends KubernetesLeaderElector.LeaderCallbackHandler {
+
+		@Override
+		public void isLeader() {
+			onGrantLeadership();
+		}
+
+		@Override
+		public void notLeader() {
+			// Clear the leader information in ConfigMap
+			try {
+				kubeClient.checkAndUpdateConfigMap(
+					configMapName,
+					configMap -> {
+						// Do not need to check the leader here
+						configMap.getData().remove(LEADER_ADDRESS_KEY);
+						configMap.getData().remove(LEADER_SESSION_ID_KEY);
+						return Optional.of(configMap);
+					}
+				).get();
+			} catch (Exception e) {
+				leaderContender.handleError(
+					new Exception("Could not remove leader information from ConfigMap " + configMapName, e));
+			}
+			onRevokeLeadership();
+			// Continue to contend the leader
+			CompletableFuture.runAsync(leaderElector::run, executor);
+		}
+	}
+
+	private class ConfigMapCallbackHandlerImpl implements FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> {
+
+		@Override
+		public void onAdded(List<KubernetesConfigMap> configMaps) {
+			// noop
+		}
+
+		@Override
+		public void onModified(List<KubernetesConfigMap> configMaps) {

Review comment:
       How can it happen that the underlying configMap gets modified? Shouldn't only the leader be able to write new information by checking that the `LEADER_ANNOTATION_KEY` contains his `LOCK_IDENTITY` and only write if this is the case?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
##########
@@ -104,6 +106,67 @@ KubernetesWatch watchPodsAndDoCallback(
 		Map<String, String> labels,
 		WatchCallbackHandler<KubernetesPod> podCallbackHandler);
 
+	/**
+	 * Create the ConfigMap with specified content. If the ConfigMap already exists, a FlinkRuntimeException will be
+	 * thrown.
+	 *
+	 * @param configMap ConfigMap.
+	 *
+	 * @return Return the ConfigMap create future.
+	 */
+	CompletableFuture<Void> createConfigMap(KubernetesConfigMap configMap);
+
+	/**
+	 * Get the ConfigMap with specified name.
+	 *
+	 * @param name ConfigMap name.
+	 *
+	 * @return Return the ConfigMap, or empty if the ConfigMap does not exist.
+	 */
+	Optional<KubernetesConfigMap> getConfigMap(String name);
+
+	/**
+	 * Update an existing ConfigMap with the data. Benefit from <a href=https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions>
+	 * resource version</a> and combined with {@link #getConfigMap(String)}, we could perform a get-check-and-update
+	 * transactional operation. Since concurrent modification could happen on a same ConfigMap,
+	 * the update operation may fail. We need to retry internally. The max retry attempts could be
+	 * configured via {@link org.apache.flink.kubernetes.configuration.KubernetesConfigOptions#KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES}.
+	 *
+	 * @param configMapName ConfigMap to be replaced with.
+	 * @param function      Function to be applied to the obtained ConfigMap and get a new updated one. If the returned

Review comment:
       What happens if the function throws an exception? I am asking because we are using a `FunctionWithException` here.

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalServiceTest.java
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.utils.Constants;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link KubernetesLeaderRetrievalService}.
+ */
+public class KubernetesLeaderRetrievalServiceTest extends KubernetesHighAvailabilityTestBase {

Review comment:
       What about integration tests for the `KubernetesLeaderElectionService` and the `KubernetesLeaderRetrievalService` with some K8s server?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.resources;
+
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfigBuilder;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
+import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+
+/**
+ * Represent {@link KubernetesLeaderElector} in kubernetes. {@link LeaderElector#run()} is a blocking call. It should be
+ *  run in the IO executor, not the main thread. The lifecycle is bound to single leader election. Once the leadership
+ * is revoked, as well as the {@link LeaderCallbackHandler#notLeader()} is called, the {@link LeaderElector#run()} will
+ * finish. To start another round of election, we need to trigger again.
+ */
+public class KubernetesLeaderElector extends LeaderElector<NamespacedKubernetesClient> {
+
+	private static final Logger LOG = LoggerFactory.getLogger(KubernetesLeaderElector.class);
+	protected static final String LOCK_IDENTITY = UUID.randomUUID().toString();
+	protected static final String LEADER_ANNOTATION_KEY = "control-plane.alpha.kubernetes.io/leader";
+
+	public KubernetesLeaderElector(
+			NamespacedKubernetesClient kubernetesClient,
+			String namespace,
+			KubernetesLeaderElectionConfiguration leaderConfig,
+			LeaderCallbackHandler leaderCallbackHandler) {
+		super(kubernetesClient, new LeaderElectionConfigBuilder()
+			.withName(leaderConfig.getConfigMapName())
+			.withLeaseDuration(leaderConfig.getLeaseDuration())
+			.withLock(new ConfigMapLock(namespace, leaderConfig.getConfigMapName(), LOCK_IDENTITY))
+			.withRenewDeadline(leaderConfig.getRenewDeadline())
+			.withRetryPeriod(leaderConfig.getRetryPeriod())
+			.withLeaderCallbacks(new LeaderCallbacks(
+				leaderCallbackHandler::isLeader,
+				leaderCallbackHandler::notLeader,
+				newLeader -> LOG.info("New leader elected {}.", newLeader)
+			))
+			.build());
+		LOG.info("Create KubernetesLeaderElector {} with lock identity {}.",
+			leaderConfig.getConfigMapName(), LOCK_IDENTITY);
+	}
+
+	public boolean hasLeadership(KubernetesConfigMap configMap) {
+		final String leader = configMap.getAnnotations().get(LEADER_ANNOTATION_KEY);

Review comment:
       Could you explain in the JavaDocs, how the `LEADER_ANNOTATION_KEY` becomes an annotation of the `configMap`? And maybe add a bit of details which underlying K8s principles we are using here.

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionServiceTest.java
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.runtime.leaderelection.TestingContender;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_KEY;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link KubernetesLeaderElectionService}.
+ */
+public class KubernetesLeaderElectionServiceTest extends KubernetesHighAvailabilityTestBase {
+
+	@Test
+	public void testKubernetesLeaderElection() throws Exception {
+		new Context() {{
+			runTestAndGrantLeadershipToContender(
+				() -> {
+					assertThat(leaderElectionService.getLeaderSessionID(), is(contender.getLeaderSessionID()));
+					assertThat(configMapStore.size(), is(1));
+					assertThat(getLeaderConfigMap().getData().get(Constants.LEADER_ADDRESS_KEY), is(LEADER_URL));
+
+					// Revoke leader
+					leaderController.set(false);
+					contender.waitForRevokeLeader(TIMEOUT);
+					assertThat(leaderElectionService.getLeaderSessionID(), nullValue());
+					assertThat(getLeaderConfigMap().getData().size(), is(0));
+				});
+		}};
+	}
+
+	@Test
+	public void testLeaderConfigMapDeletedExternally() throws Exception {
+		new Context() {{
+			runTestAndGrantLeadershipToContender(
+				() -> {
+					final FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> callbackHandler =
+						configMapsAndDoCallbackFuture.get(TIMEOUT, TimeUnit.MILLISECONDS);
+					assertThat(callbackHandler, is(notNullValue()));
+
+					callbackHandler.onDeleted(Collections.singletonList(configMapStore.remove(LEADER_CONFIGMAP_NAME)));
+					// The ConfigMap should be created again.
+					assertThat(getLeaderConfigMap().getData().get(Constants.LEADER_ADDRESS_KEY), is(LEADER_URL));
+				});
+		}};
+	}

Review comment:
       Is this something we want and have to support?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/KubeClientFactory.java
##########
@@ -71,7 +73,12 @@ public static FlinkKubeClient fromConfiguration(Configuration flinkConfig) {
 
 		final KubernetesClient client = new DefaultKubernetesClient(config);
 
-		return new Fabric8FlinkKubeClient(flinkConfig, client, KubeClientFactory::createThreadPoolForAsyncIO);
+		if (flinkKubeClient == null) {
+			flinkKubeClient = new Fabric8FlinkKubeClient(
+				flinkConfig, client, KubeClientFactory::createThreadPoolForAsyncIO);
+		}

Review comment:
       How many `flinkKubeClients` will be created when running Flink on K8s?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderElectionService{configMapName='" + configMapName + "'}";
+	}
+
+	private class LeaderCallbackHandlerImpl extends KubernetesLeaderElector.LeaderCallbackHandler {
+
+		@Override
+		public void isLeader() {
+			onGrantLeadership();
+		}
+
+		@Override
+		public void notLeader() {
+			// Clear the leader information in ConfigMap
+			try {
+				kubeClient.checkAndUpdateConfigMap(
+					configMapName,
+					configMap -> {
+						// Do not need to check the leader here
+						configMap.getData().remove(LEADER_ADDRESS_KEY);
+						configMap.getData().remove(LEADER_SESSION_ID_KEY);
+						return Optional.of(configMap);
+					}
+				).get();
+			} catch (Exception e) {
+				leaderContender.handleError(
+					new Exception("Could not remove leader information from ConfigMap " + configMapName, e));
+			}
+			onRevokeLeadership();
+			// Continue to contend the leader
+			CompletableFuture.runAsync(leaderElector::run, executor);
+		}
+	}
+
+	private class ConfigMapCallbackHandlerImpl implements FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> {
+
+		@Override
+		public void onAdded(List<KubernetesConfigMap> configMaps) {
+			// noop
+		}
+
+		@Override
+		public void onModified(List<KubernetesConfigMap> configMaps) {
+			if (hasLeadership()) {
+				configMaps.forEach(configMap -> {
+					if (isLeaderUpdatedExternally(configMap)) {
+						if (configMap.getName().equals(configMapName)) {
+							// the data field does not correspond to the expected leader information
+							if (logger.isDebugEnabled()) {
+								logger.debug("Correcting leader information in {} by {}.",
+									configMapName, leaderContender.getDescription());
+							}
+							writeLeaderInformation();
+						} else {
+							logger.warn("Ignoring the modified event since it does not belong to {}.", this);
+						}
+					}
+				});
+			}
+		}
+
+		@Override
+		public void onDeleted(List<KubernetesConfigMap> configMaps) {

Review comment:
       How can it happen that ConfigMaps are being deleted?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionService.java
##########
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesWatch;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.runtime.leaderelection.AbstractLeaderElectionService;
+import org.apache.flink.runtime.leaderelection.LeaderContender;
+import org.apache.flink.util.function.FunctionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Leader election service for multiple JobManagers. The active JobManager is elected using Kubernetes.
+ * The current leader's address as well as its leader session ID is published via Kubernetes ConfigMap.
+ * Note that the contending lock and leader storage are using the same ConfigMap. And every component(e.g.
+ * ResourceManager, Dispatcher, RestEndpoint, JobManager for each job) will have a separate ConfigMap.
+ */
+public class KubernetesLeaderElectionService extends AbstractLeaderElectionService {
+
+	private final FlinkKubeClient kubeClient;
+
+	private final Executor executor;
+
+	private final String configMapName;
+
+	private final KubernetesLeaderElector leaderElector;
+
+	private KubernetesWatch kubernetesWatch;
+
+	// Labels will be used to clean up the ha related ConfigMaps.
+	private Map<String, String> configMapLabels;
+
+	KubernetesLeaderElectionService(
+			FlinkKubeClient kubeClient,
+			Executor executor,
+			KubernetesLeaderElectionConfiguration leaderConfig) {
+
+		this.kubeClient = checkNotNull(kubeClient, "Kubernetes client should not be null.");
+		this.executor = checkNotNull(executor, "Executor should not be null.");
+		this.configMapName = leaderConfig.getConfigMapName();
+		this.leaderElector = kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+		this.leaderContender = null;
+		this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+			leaderConfig.getClusterId(), LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+	}
+
+	@Override
+	public void internalStart(LeaderContender contender) {
+		CompletableFuture.runAsync(leaderElector::run, executor);
+		kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new ConfigMapCallbackHandlerImpl());
+	}
+
+	@Override
+	public void internalStop() {
+		if (kubernetesWatch != null) {
+			kubernetesWatch.close();
+		}
+	}
+
+	@Override
+	protected void writeLeaderInformation() {
+		try {
+			kubeClient.checkAndUpdateConfigMap(
+				configMapName,
+				configMap -> {
+					if (leaderElector.hasLeadership(configMap)) {
+						// Get the updated ConfigMap with new leader information
+						if (confirmedLeaderAddress != null && confirmedLeaderSessionID != null) {
+							configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+							configMap.getData().put(LEADER_SESSION_ID_KEY, confirmedLeaderSessionID.toString());
+						}
+						configMap.getLabels().putAll(configMapLabels);
+						return Optional.of(configMap);
+					}
+					return Optional.empty();
+				}).get();
+		} catch (Exception e) {
+			leaderContender.handleError(new Exception("Could not update ConfigMap " + configMapName, e));
+		}
+	}
+
+	@Override
+	protected boolean hasLeadership() {
+		return kubeClient.getConfigMap(configMapName)
+			.map(leaderElector::hasLeadership)
+			.orElse(false);
+	}
+
+	@Override
+	public String toString() {
+		return "KubernetesLeaderElectionService{configMapName='" + configMapName + "'}";
+	}
+
+	private class LeaderCallbackHandlerImpl extends KubernetesLeaderElector.LeaderCallbackHandler {
+
+		@Override
+		public void isLeader() {
+			onGrantLeadership();
+		}
+
+		@Override
+		public void notLeader() {
+			// Clear the leader information in ConfigMap
+			try {
+				kubeClient.checkAndUpdateConfigMap(
+					configMapName,
+					configMap -> {
+						// Do not need to check the leader here
+						configMap.getData().remove(LEADER_ADDRESS_KEY);
+						configMap.getData().remove(LEADER_SESSION_ID_KEY);
+						return Optional.of(configMap);
+					}
+				).get();
+			} catch (Exception e) {
+				leaderContender.handleError(
+					new Exception("Could not remove leader information from ConfigMap " + configMapName, e));
+			}
+			onRevokeLeadership();
+			// Continue to contend the leader
+			CompletableFuture.runAsync(leaderElector::run, executor);

Review comment:
       `FutureUtils.assertNoException`

##########
File path: flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionServiceTest.java
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.highavailability;
+
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.runtime.leaderelection.TestingContender;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
+import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_KEY;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link KubernetesLeaderElectionService}.
+ */
+public class KubernetesLeaderElectionServiceTest extends KubernetesHighAvailabilityTestBase {
+
+	@Test
+	public void testKubernetesLeaderElection() throws Exception {
+		new Context() {{
+			runTestAndGrantLeadershipToContender(
+				() -> {
+					assertThat(leaderElectionService.getLeaderSessionID(), is(contender.getLeaderSessionID()));
+					assertThat(configMapStore.size(), is(1));
+					assertThat(getLeaderConfigMap().getData().get(Constants.LEADER_ADDRESS_KEY), is(LEADER_URL));
+
+					// Revoke leader
+					leaderController.set(false);
+					contender.waitForRevokeLeader(TIMEOUT);
+					assertThat(leaderElectionService.getLeaderSessionID(), nullValue());
+					assertThat(getLeaderConfigMap().getData().size(), is(0));
+				});
+		}};
+	}
+
+	@Test
+	public void testLeaderConfigMapDeletedExternally() throws Exception {
+		new Context() {{
+			runTestAndGrantLeadershipToContender(
+				() -> {
+					final FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> callbackHandler =
+						configMapsAndDoCallbackFuture.get(TIMEOUT, TimeUnit.MILLISECONDS);
+					assertThat(callbackHandler, is(notNullValue()));
+
+					callbackHandler.onDeleted(Collections.singletonList(configMapStore.remove(LEADER_CONFIGMAP_NAME)));
+					// The ConfigMap should be created again.
+					assertThat(getLeaderConfigMap().getData().get(Constants.LEADER_ADDRESS_KEY), is(LEADER_URL));
+				});
+		}};
+	}
+
+	@Test
+	public void testLeaderConfigMapUpdatedExternally() throws Exception {
+		new Context() {{
+			runTestAndGrantLeadershipToContender(
+				() -> {
+					final FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> callbackHandler =
+						configMapsAndDoCallbackFuture.get(TIMEOUT, TimeUnit.MILLISECONDS);
+					assertThat(callbackHandler, is(notNullValue()));
+					// Update ConfigMap with wrong data
+					getLeaderConfigMap().getData().put(Constants.LEADER_ADDRESS_KEY, "wrong data");
+					callbackHandler.onModified(Collections.singletonList(configMapStore.get(LEADER_CONFIGMAP_NAME)));
+					// The ConfigMap should be corrected
+					assertThat(getLeaderConfigMap().getData().get(Constants.LEADER_ADDRESS_KEY), is(LEADER_URL));
+				});
+		}};
+	}
+
+	/**
+	 * Start multiple leaders, if current leader dies, a new one could take over and update the leader
+	 * information successfully.
+	 */
+	@Test
+	public void testMultipleLeaders() throws Exception {
+		final int leaderNum = 3;
+		final AtomicBoolean[] leaderController = new AtomicBoolean[leaderNum];
+		final TestingContender[] contenders = new TestingContender[leaderNum];
+		final KubernetesLeaderElectionService[] leaderElectionServices = new KubernetesLeaderElectionService[leaderNum];
+		for (int i = 0; i < leaderNum; i++) {
+			leaderController[i] = new AtomicBoolean(false);
+			leaderElectionServices[i] = new Context().createLeaderElectionService(leaderController[i]);
+			contenders[i] = new TestingContender(getLeaderUrl(i), leaderElectionServices[i]);
+			leaderElectionServices[i].start(contenders[i]);
+		}
+		leaderController[0].set(true);
+		contenders[0].waitForLeader(TIMEOUT);
+		assertThat(
+			configMapStore.get(LEADER_CONFIGMAP_NAME).getData().get(Constants.LEADER_ADDRESS_KEY),
+			is(getLeaderUrl(0)));
+		// Leader 0 died
+		leaderController[0].set(false);
+		contenders[0].waitForRevokeLeader(TIMEOUT);
+		// Leader 2 try to acquire
+		leaderController[2].set(true);
+		contenders[2].waitForLeader(TIMEOUT);
+		assertThat(
+			configMapStore.get(LEADER_CONFIGMAP_NAME).getData().get(Constants.LEADER_ADDRESS_KEY),
+			is(getLeaderUrl(2)));
+		for (int i = 0; i < leaderNum; i++) {
+			leaderElectionServices[i].stop();
+		}
+	}

Review comment:
       Shouldn't we also have a test were we have multiple leaders concurrently trying to obtain the leadership using the proper `KubernetesLeaderElector`?

##########
File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.resources;
+
+import org.apache.flink.kubernetes.kubeclient.KubernetesLeaderElectionConfiguration;
+
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfigBuilder;
+import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
+import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+
+/**
+ * Represent {@link KubernetesLeaderElector} in kubernetes. {@link LeaderElector#run()} is a blocking call. It should be
+ *  run in the IO executor, not the main thread. The lifecycle is bound to single leader election. Once the leadership
+ * is revoked, as well as the {@link LeaderCallbackHandler#notLeader()} is called, the {@link LeaderElector#run()} will
+ * finish. To start another round of election, we need to trigger again.
+ */
+public class KubernetesLeaderElector extends LeaderElector<NamespacedKubernetesClient> {

Review comment:
       I think we are missing tests for this class.




----------------------------------------------------------------
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:
users@infra.apache.org