wangyang0918 commented on a change in pull request #13644:
URL: https://github.com/apache/flink/pull/13644#discussion_r516404322
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/FutureUtilsTest.java
##########
@@ -160,6 +161,39 @@ public void testRetryCancellation() throws Exception {
}
}
+ /**
+ * Test that {@link FutureUtils#retry} should stop at non-retryable
exception.
+ */
+ @Test
+ public void testStopAtNonRetryableException() {
+ final int retries = 10;
+ final int notRetry = 3;
+ final AtomicInteger atomicInteger = new AtomicInteger(0);
+ final String notRetryExceptionMsg = "Non-retryable exception";
+ CompletableFuture<Boolean> retryFuture = FutureUtils.retry(
+ () -> CompletableFuture.supplyAsync(
+ () -> {
+ if (atomicInteger.incrementAndGet() ==
notRetry) {
+ // throw non-retryable exception
+ throw new
CompletionException(new FlinkRuntimeException(notRetryExceptionMsg));
+ } else {
+ throw new
CompletionException(new FlinkException("Test exception"));
+ }
+ },
+ TestingUtils.defaultExecutor()),
+ retries,
+ throwable -> ExceptionUtils.findThrowable(throwable,
FlinkException.class).isPresent(),
+ TestingUtils.defaultExecutor());
+
+ try {
+ retryFuture.get();
+ fail("Exception should be thrown.");
+ } catch (Exception ex) {
+ assertThat(ExceptionUtils.findThrowableWithMessage(ex,
notRetryExceptionMsg).isPresent(), is(true));
Review comment:
Nice suggestion. I will change other similar exception `assert`.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.Objects;
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation for leader election service. Composed with different
{@link LeaderElectionDriver}, we could
+ * perform a leader election for the contender, and then persist the leader
information to various storage.
+ */
+public class DefaultLeaderElectionService implements LeaderElectionService,
LeaderElectionEventHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DefaultLeaderElectionService.class);
+
+ private final Object lock = new Object();
+
+ private final LeaderElectionDriverFactory leaderElectionDriverFactory;
+
+ /** The leader contender which applies for leadership. */
+ @GuardedBy("lock")
+ private volatile LeaderContender leaderContender;
+
+ @GuardedBy("lock")
+ private volatile UUID issuedLeaderSessionID;
+
+ @GuardedBy("lock")
+ private volatile UUID confirmedLeaderSessionID;
+
+ @GuardedBy("lock")
+ private volatile String confirmedLeaderAddress;
+
+ @GuardedBy("lock")
+ private volatile boolean running;
+
+ @GuardedBy("lock")
+ private LeaderElectionDriver leaderElectionDriver;
+
+ public DefaultLeaderElectionService(LeaderElectionDriverFactory
leaderElectionDriverFactory) {
+ this.leaderElectionDriverFactory =
checkNotNull(leaderElectionDriverFactory);
+
+ leaderContender = null;
+
+ issuedLeaderSessionID = null;
+ confirmedLeaderSessionID = null;
+ confirmedLeaderAddress = null;
+
+ running = false;
+ }
+
+ @Override
+ public final void start(LeaderContender contender) throws Exception {
+ checkNotNull(contender, "Contender must not be null.");
+ Preconditions.checkState(leaderContender == null, "Contender
was already set.");
+
+ synchronized (lock) {
+ leaderContender = contender;
+ leaderElectionDriver =
leaderElectionDriverFactory.createLeaderElectionDriver(
+ this, new LeaderElectionFatalErrorHandler(),
leaderContender.getDescription());
+ LOG.info("Starting DefaultLeaderElectionService with
{}.", leaderElectionDriver);
+
+ running = true;
+ }
+ }
+
+ @Override
+ public final void stop() throws Exception {
+ LOG.info("Stopping DefaultLeaderElectionService.");
+
+ synchronized (lock) {
+ if (!running) {
+ return;
+ }
+ running = false;
+ clearConfirmedLeaderInformation();
+ leaderElectionDriver.close();
+ }
+ }
+
+ @Override
+ public void confirmLeadership(UUID leaderSessionID, String
leaderAddress) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Confirm leader session ID {} for leader {}.",
+ leaderSessionID,
+ leaderAddress);
+ }
+
+ checkNotNull(leaderSessionID);
+
+ synchronized (lock) {
+ if (hasLeadership(leaderSessionID)) {
+ if (running) {
+
confirmLeaderInformation(leaderSessionID, leaderAddress);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring the leader
session Id {} confirmation, since the " +
+ "LeaderElectionService
has already been stopped.", leaderSessionID);
+ }
+ }
+ } else {
+ // Received an old confirmation call
+ if
(!leaderSessionID.equals(this.issuedLeaderSessionID)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Receive an old
confirmation call of leader session ID {}, " +
+ "current issued session
ID is {}", leaderSessionID, issuedLeaderSessionID);
+ }
+ } else {
+ LOG.warn("The leader session ID {} was
confirmed even though the " +
+ "corresponding JobManager was
not elected as the leader.", leaderSessionID);
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean hasLeadership(@Nonnull UUID leaderSessionId) {
+ synchronized (lock) {
+ if (running) {
+ return leaderElectionDriver.hasLeadership() &&
leaderSessionId.equals(issuedLeaderSessionID);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("hasLeadership is called
after the service is stopped, returning false.");
+ }
+ return false;
+ }
+ }
+ }
+
+ /**
+ * 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
+ */
+ @VisibleForTesting
+ @Nullable
+ public UUID getLeaderSessionID() {
+ return confirmedLeaderSessionID;
+ }
+
+ @GuardedBy("lock")
+ private void confirmLeaderInformation(UUID leaderSessionID, String
leaderAddress) {
+ confirmedLeaderSessionID = leaderSessionID;
+ confirmedLeaderAddress = leaderAddress;
+ leaderElectionDriver.writeLeaderInformation(
+ LeaderInformation.known(confirmedLeaderSessionID,
confirmedLeaderAddress));
+ }
+
+ @GuardedBy("lock")
+ private void clearConfirmedLeaderInformation() {
+ confirmedLeaderSessionID = null;
+ confirmedLeaderAddress = null;
+ }
+
+ @Override
+ @GuardedBy("lock")
+ public void onGrantLeadership() {
+ synchronized (lock) {
+ if (running) {
+ issuedLeaderSessionID = UUID.randomUUID();
+ clearConfirmedLeaderInformation();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Grant leadership to contender
{} with session ID {}.",
+
leaderContender.getDescription(),
+ issuedLeaderSessionID);
+ }
+
+
leaderContender.grantLeadership(issuedLeaderSessionID);
Review comment:
Yes. We may have a potential deadlock here. And callback of
`leaderContender` should not happen under the `lock`. I will move all the
callback out of `lock`.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriver.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * A {@link LeaderElectionDriver} is responsible for performing the leader
election and storing the leader information.
+ * All the leader internal state is guarded by lock in {@link
LeaderElectionService}. Different driver
+ * implementations do not need to care about the lock. And it should use
{@link LeaderElectionEventHandler}
+ * if it want to respond to the leader change events.
+ *
+ * <p><strong>Important</strong>: The {@link LeaderElectionDriver} could not
guarantee that there is no
+ * {@link LeaderElectionEventHandler} callbacks happen after {@link #close()}.
+ */
+public interface LeaderElectionDriver {
+
+ /**
+ * Write the current leader information to external persistent
storage(e.g. Zookeeper, Kubernetes ConfigMap). This
+ * is a blocking IO operation.
+ *
+ * @param leaderInformation current leader information. It could be
{@link LeaderInformation#empty()}, which means
+ * the caller want to clear the leader information on external storage.
+ */
+ void writeLeaderInformation(LeaderInformation leaderInformation);
Review comment:
`writeLeaderInformation` will only take effect when the current driver
still have the leadership. The caller do not need to make sure of it. I will
add this to the interface description.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/LeaderElectionDriver.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * A {@link LeaderElectionDriver} is responsible for performing the leader
election and storing the leader information.
+ * All the leader internal state is guarded by lock in {@link
LeaderElectionService}. Different driver
+ * implementations do not need to care about the lock. And it should use
{@link LeaderElectionEventHandler}
+ * if it want to respond to the leader change events.
+ *
+ * <p><strong>Important</strong>: The {@link LeaderElectionDriver} could not
guarantee that there is no
+ * {@link LeaderElectionEventHandler} callbacks happen after {@link #close()}.
+ */
+public interface LeaderElectionDriver {
+
+ /**
+ * Write the current leader information to external persistent
storage(e.g. Zookeeper, Kubernetes ConfigMap). This
+ * is a blocking IO operation.
+ *
+ * @param leaderInformation current leader information. It could be
{@link LeaderInformation#empty()}, which means
+ * the caller want to clear the leader information on external storage.
+ */
+ void writeLeaderInformation(LeaderInformation leaderInformation);
Review comment:
"The write operation takes effect only when the driver still has the
leadership." The caller do not need to make sure of it. I will add this to the
interface description.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriver.java
##########
@@ -0,0 +1,291 @@
+/*
+ * 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.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.util.ExceptionUtils;
+
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.api.UnhandledErrorListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.ChildData;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCache;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCacheListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.leader.LeaderLatch;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.leader.LeaderLatchListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.state.ConnectionState;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.data.Stat;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link LeaderElectionDriver} implementation for Zookeeper. The leading
JobManager is elected using
+ * ZooKeeper. The current leader's address as well as its leader session ID is
published via
+ * ZooKeeper.
+ */
+public class ZooKeeperLeaderElectionDriver implements LeaderElectionDriver,
LeaderLatchListener, NodeCacheListener, UnhandledErrorListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ZooKeeperLeaderElectionDriver.class);
+
+ /** Client to the ZooKeeper quorum. */
+ private final CuratorFramework client;
+
+ /** Curator recipe for leader election. */
+ private final LeaderLatch leaderLatch;
+
+ /** Curator recipe to watch a given ZooKeeper node for changes. */
+ private final NodeCache cache;
+
+ /** ZooKeeper path of the node which stores the current leader
information. */
+ private final String leaderPath;
+
+ private final ConnectionStateListener listener = (client, newState) ->
handleStateChange(newState);
+
+ private final LeaderElectionEventHandler leaderElectionEventHandler;
+
+ private final FatalErrorHandler fatalErrorHandler;
+
+ private final String leaderContenderDescription;
+
+ private volatile boolean running;
+
+ /**
+ * Creates a ZooKeeperLeaderElectionDriver object.
+ *
+ * @param client Client which is connected to the ZooKeeper quorum
+ * @param latchPath ZooKeeper node path for the leader election latch
+ * @param leaderPath ZooKeeper node path for the node which stores the
current leader information
+ * @param leaderElectionEventHandler Event handler for processing
leader change events
+ * @param fatalErrorHandler Fatal error handler
+ * @param leaderContenderDescription Leader contender description
+ */
+ public ZooKeeperLeaderElectionDriver(
+ CuratorFramework client,
+ String latchPath,
+ String leaderPath,
+ LeaderElectionEventHandler leaderElectionEventHandler,
+ FatalErrorHandler fatalErrorHandler,
+ String leaderContenderDescription) throws Exception {
+ this.client = checkNotNull(client);
+ this.leaderPath = checkNotNull(leaderPath);
+ this.leaderElectionEventHandler =
checkNotNull(leaderElectionEventHandler);
+ this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
+ this.leaderContenderDescription =
checkNotNull(leaderContenderDescription);
+
+ leaderLatch = new LeaderLatch(client, checkNotNull(latchPath));
+ cache = new NodeCache(client, leaderPath);
+
+ client.getUnhandledErrorListenable().addListener(this);
+
+ leaderLatch.addListener(this);
+ leaderLatch.start();
+
+ cache.getListenable().addListener(this);
+ cache.start();
+
+ client.getConnectionStateListenable().addListener(listener);
+
+ running = true;
+ }
+
+ @Override
+ public void close() throws Exception{
+ if (!running) {
+ return;
+ }
+ running = false;
+
+ LOG.info("Closing {}", this);
+
+ client.getUnhandledErrorListenable().removeListener(this);
+
+ client.getConnectionStateListenable().removeListener(listener);
+
+ Exception exception = null;
+
+ try {
+ cache.close();
+ } catch (Exception e) {
+ exception = e;
+ }
+
+ try {
+ leaderLatch.close();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e,
exception);
+ }
+
+ if (exception != null) {
+ throw new Exception("Could not properly stop the
ZooKeeperLeaderElectionDriver.", exception);
+ }
+ }
+
+ @Override
+ public boolean hasLeadership() {
+ assert(running);
+ return leaderLatch.hasLeadership();
+ }
+
+ @Override
+ public void isLeader() {
+ leaderElectionEventHandler.onGrantLeadership();
+ }
+
+ @Override
+ public void notLeader() {
+ leaderElectionEventHandler.onRevokeLeadership();
+ }
+
+ @Override
+ public void nodeChanged() throws Exception {
+ if (leaderLatch.hasLeadership()) {
+ ChildData childData = cache.getCurrentData();
+ if (childData != null) {
+ final byte[] data = childData.getData();
+ if (data != null && data.length > 0) {
+ final ByteArrayInputStream bais = new
ByteArrayInputStream(data);
+ final ObjectInputStream ois = new
ObjectInputStream(bais);
+
+ final String leaderAddress =
ois.readUTF();
+ final UUID leaderSessionID = (UUID)
ois.readObject();
+
+
leaderElectionEventHandler.onLeaderInformationChange(
+
LeaderInformation.known(leaderSessionID, leaderAddress));
+ return;
+ }
+ }
+
leaderElectionEventHandler.onLeaderInformationChange(LeaderInformation.empty());
+ }
+ }
+
+ /**
+ * Writes the current leader's address as well the given leader session
ID to ZooKeeper.
+ */
+ @Override
+ public void writeLeaderInformation(LeaderInformation leaderInformation)
{
+ assert(running);
+ // this method does not have to be synchronized because the
curator framework client
+ // is thread-safe. We do not write the empty data to ZooKeeper
here. Because check-leadership-and-update
+ // is not a transactional operation. We may wrongly clear the
data written by new leader.
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Write leader information: {}.",
leaderInformation);
+ }
+ if (leaderInformation.equals(LeaderInformation.empty())) {
+ return;
Review comment:
I think the normal behavior should be that once the current
`LeaderElectionDriver` is not the leader, it is responsible for cleaning up the
leader information on the external storage(Kubernetes, ZooKeeper). And it
should only be executed when a new leader has not write his leader information.
For ZooKeeper, we are using an ephemeral node for leader information and it
will be cleaned up automatically when ZooKeeper client timeout. So it is
reasonable to not do the clean up here.
However, for Kubernetes, we should and could do this. Since check the
leadership(aka check the annotation on the leader ConfigMap) and remove leader
information is a transactional operation, we could guarantee that we will not
overwrite a new leader information.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionDriver.java
##########
@@ -0,0 +1,291 @@
+/*
+ * 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.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.util.ExceptionUtils;
+
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.CuratorFramework;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.api.UnhandledErrorListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.ChildData;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCache;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.cache.NodeCacheListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.leader.LeaderLatch;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.recipes.leader.LeaderLatchListener;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.state.ConnectionState;
+import
org.apache.flink.shaded.curator4.org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
+import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.data.Stat;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link LeaderElectionDriver} implementation for Zookeeper. The leading
JobManager is elected using
+ * ZooKeeper. The current leader's address as well as its leader session ID is
published via
+ * ZooKeeper.
+ */
+public class ZooKeeperLeaderElectionDriver implements LeaderElectionDriver,
LeaderLatchListener, NodeCacheListener, UnhandledErrorListener {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ZooKeeperLeaderElectionDriver.class);
+
+ /** Client to the ZooKeeper quorum. */
+ private final CuratorFramework client;
+
+ /** Curator recipe for leader election. */
+ private final LeaderLatch leaderLatch;
+
+ /** Curator recipe to watch a given ZooKeeper node for changes. */
+ private final NodeCache cache;
+
+ /** ZooKeeper path of the node which stores the current leader
information. */
+ private final String leaderPath;
+
+ private final ConnectionStateListener listener = (client, newState) ->
handleStateChange(newState);
+
+ private final LeaderElectionEventHandler leaderElectionEventHandler;
+
+ private final FatalErrorHandler fatalErrorHandler;
+
+ private final String leaderContenderDescription;
+
+ private volatile boolean running;
+
+ /**
+ * Creates a ZooKeeperLeaderElectionDriver object.
+ *
+ * @param client Client which is connected to the ZooKeeper quorum
+ * @param latchPath ZooKeeper node path for the leader election latch
+ * @param leaderPath ZooKeeper node path for the node which stores the
current leader information
+ * @param leaderElectionEventHandler Event handler for processing
leader change events
+ * @param fatalErrorHandler Fatal error handler
+ * @param leaderContenderDescription Leader contender description
+ */
+ public ZooKeeperLeaderElectionDriver(
+ CuratorFramework client,
+ String latchPath,
+ String leaderPath,
+ LeaderElectionEventHandler leaderElectionEventHandler,
+ FatalErrorHandler fatalErrorHandler,
+ String leaderContenderDescription) throws Exception {
+ this.client = checkNotNull(client);
+ this.leaderPath = checkNotNull(leaderPath);
+ this.leaderElectionEventHandler =
checkNotNull(leaderElectionEventHandler);
+ this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
+ this.leaderContenderDescription =
checkNotNull(leaderContenderDescription);
+
+ leaderLatch = new LeaderLatch(client, checkNotNull(latchPath));
+ cache = new NodeCache(client, leaderPath);
+
+ client.getUnhandledErrorListenable().addListener(this);
+
+ leaderLatch.addListener(this);
+ leaderLatch.start();
+
+ cache.getListenable().addListener(this);
+ cache.start();
+
+ client.getConnectionStateListenable().addListener(listener);
+
+ running = true;
+ }
+
+ @Override
+ public void close() throws Exception{
+ if (!running) {
+ return;
+ }
+ running = false;
+
+ LOG.info("Closing {}", this);
+
+ client.getUnhandledErrorListenable().removeListener(this);
+
+ client.getConnectionStateListenable().removeListener(listener);
+
+ Exception exception = null;
+
+ try {
+ cache.close();
+ } catch (Exception e) {
+ exception = e;
+ }
+
+ try {
+ leaderLatch.close();
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e,
exception);
+ }
+
+ if (exception != null) {
+ throw new Exception("Could not properly stop the
ZooKeeperLeaderElectionDriver.", exception);
+ }
+ }
+
+ @Override
+ public boolean hasLeadership() {
+ assert(running);
+ return leaderLatch.hasLeadership();
+ }
+
+ @Override
+ public void isLeader() {
+ leaderElectionEventHandler.onGrantLeadership();
+ }
+
+ @Override
+ public void notLeader() {
+ leaderElectionEventHandler.onRevokeLeadership();
+ }
+
+ @Override
+ public void nodeChanged() throws Exception {
+ if (leaderLatch.hasLeadership()) {
+ ChildData childData = cache.getCurrentData();
+ if (childData != null) {
+ final byte[] data = childData.getData();
+ if (data != null && data.length > 0) {
+ final ByteArrayInputStream bais = new
ByteArrayInputStream(data);
+ final ObjectInputStream ois = new
ObjectInputStream(bais);
+
+ final String leaderAddress =
ois.readUTF();
+ final UUID leaderSessionID = (UUID)
ois.readObject();
+
+
leaderElectionEventHandler.onLeaderInformationChange(
+
LeaderInformation.known(leaderSessionID, leaderAddress));
+ return;
+ }
+ }
+
leaderElectionEventHandler.onLeaderInformationChange(LeaderInformation.empty());
+ }
+ }
+
+ /**
+ * Writes the current leader's address as well the given leader session
ID to ZooKeeper.
+ */
+ @Override
+ public void writeLeaderInformation(LeaderInformation leaderInformation)
{
+ assert(running);
+ // this method does not have to be synchronized because the
curator framework client
+ // is thread-safe. We do not write the empty data to ZooKeeper
here. Because check-leadership-and-update
+ // is not a transactional operation. We may wrongly clear the
data written by new leader.
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Write leader information: {}.",
leaderInformation);
+ }
+ if (leaderInformation.equals(LeaderInformation.empty())) {
+ return;
Review comment:
I will add more description on the interface
`LeaderElectionDriver#writeLeaderInformation`.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
##########
@@ -520,6 +519,47 @@ public void testEphemeralZooKeeperNodes() throws Exception
{
}
}
+ @Test
+ public void testNotLeader() throws Exception {
+
+ final TestingLeaderElectionEventHandler electionEventHandler =
+ new TestingLeaderElectionEventHandler(TEST_LEADER);
+ final TestingLeaderRetrievalEventHandler retrievalEventHandler
= new TestingLeaderRetrievalEventHandler();
+ LeaderElectionDriver leaderElectionDriver = null;
+ LeaderRetrievalDriver leaderRetrievalDriver = null;
+ try {
+
+ leaderElectionDriver =
+
ZooKeeperUtils.createLeaderElectionDriverFactory(client, configuration)
+
.createLeaderElectionDriver(electionEventHandler,
electionEventHandler::handleError, TEST_URL);
+ electionEventHandler.init(leaderElectionDriver);
+
+ electionEventHandler.waitForLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(TEST_LEADER));
+
+ // Leader is revoked
+ ((ZooKeeperLeaderElectionDriver)
leaderElectionDriver).notLeader();
+ electionEventHandler.waitForRevokeLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(LeaderInformation.empty()));
+ // The data on ZooKeeper it not be cleared
Review comment:
Since in the currently ZooKeeper HA implementation, check the leadership
and then update the ZooKeeper node(aka clean up the leader information here) is
not a transactional operation. So we may wrongly update the ZooKeeper node with
empty data. What I have done is do nothing and the ephemeral node disappear
automatically.
I add this test to verify this behavior and make sure when the behavior
changed we could find it.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
##########
@@ -520,6 +519,47 @@ public void testEphemeralZooKeeperNodes() throws Exception
{
}
}
+ @Test
+ public void testNotLeader() throws Exception {
+
+ final TestingLeaderElectionEventHandler electionEventHandler =
+ new TestingLeaderElectionEventHandler(TEST_LEADER);
+ final TestingLeaderRetrievalEventHandler retrievalEventHandler
= new TestingLeaderRetrievalEventHandler();
+ LeaderElectionDriver leaderElectionDriver = null;
+ LeaderRetrievalDriver leaderRetrievalDriver = null;
+ try {
+
+ leaderElectionDriver =
+
ZooKeeperUtils.createLeaderElectionDriverFactory(client, configuration)
+
.createLeaderElectionDriver(electionEventHandler,
electionEventHandler::handleError, TEST_URL);
+ electionEventHandler.init(leaderElectionDriver);
+
+ electionEventHandler.waitForLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(TEST_LEADER));
+
+ // Leader is revoked
+ ((ZooKeeperLeaderElectionDriver)
leaderElectionDriver).notLeader();
+ electionEventHandler.waitForRevokeLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(LeaderInformation.empty()));
+ // The data on ZooKeeper it not be cleared
Review comment:
Since in the current ZooKeeper HA implementation, check the leadership
and then update the ZooKeeper node(aka clean up the leader information here) is
not a transactional operation. So we may wrongly update the ZooKeeper node with
empty data. What I have done is do nothing and the ephemeral node disappear
automatically.
I add this test to verify this behavior and make sure when the behavior
changed we could find it.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
##########
@@ -520,6 +519,47 @@ public void testEphemeralZooKeeperNodes() throws Exception
{
}
}
+ @Test
+ public void testNotLeader() throws Exception {
+
+ final TestingLeaderElectionEventHandler electionEventHandler =
+ new TestingLeaderElectionEventHandler(TEST_LEADER);
+ final TestingLeaderRetrievalEventHandler retrievalEventHandler
= new TestingLeaderRetrievalEventHandler();
+ LeaderElectionDriver leaderElectionDriver = null;
+ LeaderRetrievalDriver leaderRetrievalDriver = null;
+ try {
+
+ leaderElectionDriver =
+
ZooKeeperUtils.createLeaderElectionDriverFactory(client, configuration)
+
.createLeaderElectionDriver(electionEventHandler,
electionEventHandler::handleError, TEST_URL);
+ electionEventHandler.init(leaderElectionDriver);
+
+ electionEventHandler.waitForLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(TEST_LEADER));
+
+ // Leader is revoked
+ ((ZooKeeperLeaderElectionDriver)
leaderElectionDriver).notLeader();
+ electionEventHandler.waitForRevokeLeader(timeout);
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(LeaderInformation.empty()));
+ // The data on ZooKeeper it not be cleared
Review comment:
Since in the current ZooKeeper HA implementation, check the leadership
and then update the ZooKeeper node(aka clean up the leader information here) is
not a transactional operation. So we may wrongly update the ZooKeeper node with
empty data. What I have done is doing nothing and the ephemeral node disappear
automatically.
I add this test to verify this behavior and make sure when the behavior
changed we could find it.
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientITCase.java
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.kubernetes.KubernetesResource;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * IT Tests for {@link
org.apache.flink.kubernetes.kubeclient.Fabric8FlinkKubeClient} with real K8s
server and client.
+ */
+public class Fabric8FlinkKubeClientITCase {
+
+ @ClassRule
+ public static KubernetesResource kubernetesResource = new
KubernetesResource();
+
+ private static final String TEST_CONFIG_MAP_NAME = "test-config-map";
+
+ private static final long TIMEOUT = 120L * 1000L;
+
+ private static final Map<String, String> data = new HashMap<String,
String>() {
+ {
+ put("key1", "0");
+ put("key2", "0");
+ put("key3", "0");
+ }
+ };
+
+ private FlinkKubeClient flinkKubeClient;
+
+ @Before
+ public void setup() throws Exception {
+ flinkKubeClient = kubernetesResource.getFlinkKubeClient();
+ flinkKubeClient.createConfigMap(new KubernetesConfigMap(
+ new ConfigMapBuilder()
+ .withNewMetadata()
+ .withName(TEST_CONFIG_MAP_NAME)
+ .endMetadata()
+ .withData(data)
+ .build())).get();
+ }
+
+ @After
+ public void teardown() throws Exception {
+ flinkKubeClient.deleteConfigMap(TEST_CONFIG_MAP_NAME).get();
+ }
+
+ /**
+ * {@link
org.apache.flink.kubernetes.kubeclient.FlinkKubeClient#checkAndUpdateConfigMap}
is a transactional
+ * operation, we should definitely guarantee that the concurrent
modification could work.
+ */
+ @Test
+ public void testCheckAndUpdateConfigMapConcurrently() throws Exception {
+ // Start multiple instances to update ConfigMap concurrently
+ final List<CompletableFuture<Void>> futures = new ArrayList<>();
+ final int target = 10;
+ final int updateIntervalMs = 100;
+ for (String key : data.keySet()) {
+
futures.add(FutureUtils.runAfterwardsAsync(FutureUtils.completedVoidFuture(),
() -> {
Review comment:
Because we assume to increase the number one by one and check the target
finally. Then we have to call `flinkKubeClient.checkAndUpdateConfigMap().get`
and it will throw exception here. We need to execute a `RunnableWithException`
not `Runnable`. Right?
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
##########
@@ -249,4 +271,131 @@ 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 testCreateConfigMapAlreadyExisting() throws Exception {
+ final KubernetesConfigMap configMap = buildTestingConfigMap();
+ this.flinkKubeClient.createConfigMap(configMap).get();
+
+
mockCreateConfigMapAlreadyExisting(configMap.getInternalResource());
+ configMap.getData().put(TESTING_CONFIG_MAP_KEY,
TESTING_CONFIG_MAP_NEW_VALUE);
+ try {
+ this.flinkKubeClient.createConfigMap(configMap).get();
+ fail("Exception should be thrown.");
+ } catch (Exception ex) {
+ assertThat(ex.getMessage(),
+ containsString("Failed to create ConfigMap " +
TESTING_CONFIG_MAP_NAME));
Review comment:
I will fix all the `assert` with exception in this pattern in this PR
and following PRs.
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
##########
@@ -0,0 +1,102 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import
org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.runtime.util.ExecutorThreadFactory;
+
+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.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * 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.
+ *
+ * <p>{@link LeaderElector#run()} is responsible for creating the leader
ConfigMap and continuously update the
+ * annotation. The annotation key is {@link #LEADER_ANNOTATION_KEY} and the
value is in the following json format.
+ * metadata:
+ * annotations:
+ * control-plane.alpha.kubernetes.io/leader:
'{"holderIdentity":"623e39fb-70c3-44f1-811f-561ec4a28d75","leaseDuration":15.000000000,"acquireTime":"2020-10-20T04:06:31.431000Z","renewTime":"2020-10-22T08:51:36.843000Z","leaderTransitions":37981}'
+ */
+public class KubernetesLeaderElector extends
LeaderElector<NamespacedKubernetesClient> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesLeaderElector.class);
+ @VisibleForTesting
+ public static final String LEADER_ANNOTATION_KEY =
"control-plane.alpha.kubernetes.io/leader";
+
+ private final ExecutorService executorService =
Executors.newSingleThreadExecutor(
+ new
ExecutorThreadFactory("KubernetesLeaderElector-ExecutorService"));
+
+ 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(), leaderConfig.getLockIdentity()))
+ .withRenewDeadline(leaderConfig.getRenewDeadline())
+ .withRetryPeriod(leaderConfig.getRetryPeriod())
+ .withLeaderCallbacks(new LeaderCallbacks(
+ leaderCallbackHandler::isLeader,
+ leaderCallbackHandler::notLeader,
+ newLeader -> LOG.info("New leader elected {}
for {}.", newLeader, leaderConfig.getConfigMapName())
+ ))
+ .build());
+ LOG.info("Create KubernetesLeaderElector {} with lock identity
{}.",
+ leaderConfig.getConfigMapName(),
leaderConfig.getLockIdentity());
+ }
+
+ @Override
+ public void run() {
+ CompletableFuture.runAsync(super::run, executorService);
Review comment:
We do not expect to have exception in the `LeaderElector#run`.
Otherwise, it is fatal error handled by `FatalExitExceptionHandler`.
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.configuration.Configuration;
+import org.apache.flink.kubernetes.KubernetesResource;
+import
org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.DefaultKubeClientFactory;
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.KubeClientFactory;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.assertThat;
+
+/**
+ * IT Tests for the {@link KubernetesLeaderElector}. Start multiple leader
contenders currently, one should elect
+ * successfully. And if current leader dies, a new one could take over.
+ */
+public class KubernetesLeaderElectorITCase {
+
+ @ClassRule
+ public static KubernetesResource kubernetesResource = new
KubernetesResource();
+
+ private static final long TIMEOUT = 120L * 1000L;
+
+ private final KubeClientFactory kubeClientFactory = new
DefaultKubeClientFactory();
+
+ private static final String LEADER_CONFIGMAP_NAME_PREFIX =
"leader-test-cluster";
+
+ @Test
+ public void testMultipleKubernetesLeaderElectors() throws Exception {
+ final Configuration configuration =
kubernetesResource.getConfiguration();
+ final ExecutorService executorService =
kubernetesResource.getExecutorService();
+
+ final BlockingQueue<String> leaderStore = new
LinkedBlockingQueue<>();
+ final String leaderConfigMapName = LEADER_CONFIGMAP_NAME_PREFIX
+ System.currentTimeMillis();
+ final int leaderNum = 3;
+
+ final KubernetesLeaderElector[] leaderElectors = new
KubernetesLeaderElector[leaderNum];
+ // We use different Kubernetes clients for different leader
electors.
+ final FlinkKubeClient[] kubeClients = new
FlinkKubeClient[leaderNum];
+ final TestingLeaderCallbackHandler[] leaderCallbackHandlers =
new TestingLeaderCallbackHandler[leaderNum];
+
+ try {
+ for (int i = 0; i < leaderNum; i++) {
+ kubeClients[i] =
kubeClientFactory.fromConfiguration(configuration, executorService);
+ leaderCallbackHandlers[i] = new
TestingLeaderCallbackHandler(leaderStore, UUID.randomUUID().toString());
+ final KubernetesLeaderElectionConfiguration
leaderConfig = new KubernetesLeaderElectionConfiguration(
+ leaderConfigMapName,
leaderCallbackHandlers[i].getLockIdentity(), configuration);
+ leaderElectors[i] =
kubeClients[i].createLeaderElector(leaderConfig, leaderCallbackHandlers[i]);
+
+ // Start the leader electors to contend the
leader
+ leaderElectors[i].run();
+ }
+
+ // Wait for the first leader
+ final String firstLockIdentity =
leaderStore.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+
+ for (int i = 0; i < leaderNum; i++) {
+ if
(leaderCallbackHandlers[i].getLockIdentity().equals(firstLockIdentity)) {
+ // Check the callback is called.
+
assertThat(leaderCallbackHandlers[i].hasLeadership(), is(true));
+ // Current leader died
+ leaderElectors[i].stop();
Review comment:
From the following piece of codes, we could find that `notLeader` will
be called when the renew timeout of is interrupted. I will update the test to
also verify the `notLeader` is called.
```
/**
* Starts the leader election loop
*/
public void run() {
LOGGER.debug("Leader election started");
if (!acquire()) {
return;
}
leaderElectionConfig.getLeaderCallbacks().onStartLeading();
renewWithTimeout();
leaderElectionConfig.getLeaderCallbacks().onStopLeading();
}
```
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.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 java.util.concurrent.BlockingQueue;
+
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Testing implementation for {@link
KubernetesLeaderElector.LeaderCallbackHandler}.
+ */
+public class TestingLeaderCallbackHandler extends
KubernetesLeaderElector.LeaderCallbackHandler {
+
+ private final BlockingQueue<String> leaderStore;
+ private final String lockIdentity;
+ private boolean isLeader;
+
+ public TestingLeaderCallbackHandler(BlockingQueue<String> leaderStore,
String lockIdentity) {
+ this.leaderStore = leaderStore;
+ this.lockIdentity = lockIdentity;
+ }
+
+ @Override
+ public void isLeader() {
+ isLeader = true;
+ leaderStore.poll();
+ leaderStore.offer(lockIdentity);
+ assertThat(leaderStore.size(), is(1));
Review comment:
Make sense. I will make the queue private and add `waitForXXX` method.
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriver.java
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.configuration.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.KubernetesException;
+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.LeaderElectionDriver;
+import org.apache.flink.runtime.leaderelection.LeaderElectionEventHandler;
+import org.apache.flink.runtime.leaderelection.LeaderElectionException;
+import org.apache.flink.runtime.leaderelection.LeaderInformation;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+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.kubernetes.utils.KubernetesUtils.checkConfigMaps;
+import static
org.apache.flink.kubernetes.utils.KubernetesUtils.getLeaderInformationFromConfigMap;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@link LeaderElectionDriver} implementation for Kubernetes. The active
leader 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 KubernetesLeaderElectionDriver implements LeaderElectionDriver {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesLeaderElectionDriver.class);
+
+ private final FlinkKubeClient kubeClient;
+
+ private final String configMapName;
+
+ private final String lockIdentity;
+
+ private final KubernetesLeaderElector leaderElector;
+
+ // Labels will be used to clean up the ha related ConfigMaps.
+ private final Map<String, String> configMapLabels;
+
+ private final LeaderElectionEventHandler leaderElectionEventHandler;
+
+ private final KubernetesWatch kubernetesWatch;
+
+ private final FatalErrorHandler fatalErrorHandler;
+
+ private volatile boolean running;
+
+ public KubernetesLeaderElectionDriver(
+ FlinkKubeClient kubeClient,
+ KubernetesLeaderElectionConfiguration leaderConfig,
+ LeaderElectionEventHandler leaderElectionEventHandler,
+ FatalErrorHandler fatalErrorHandler) {
+
+ this.kubeClient = checkNotNull(kubeClient, "Kubernetes client");
+ checkNotNull(leaderConfig, "Leader election configuration");
+ this.leaderElectionEventHandler =
checkNotNull(leaderElectionEventHandler, "LeaderElectionEventHandler");
+ this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
+
+ this.configMapName = leaderConfig.getConfigMapName();
+ this.lockIdentity = leaderConfig.getLockIdentity();
+ this.leaderElector =
kubeClient.createLeaderElector(leaderConfig, new LeaderCallbackHandlerImpl());
+ this.configMapLabels = KubernetesUtils.getConfigMapLabels(
+ leaderConfig.getClusterId(),
LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
+
+ leaderElector.run();
+ kubernetesWatch = kubeClient.watchConfigMaps(configMapName, new
ConfigMapCallbackHandlerImpl());
+ running = true;
+ }
+
+ @Override
+ public void close() {
+ if (!running) {
+ return;
+ }
+ running = false;
+
+ LOG.info("Closing {}.", this);
+ leaderElector.stop();
+ kubernetesWatch.close();
+ }
+
+ @Override
+ public void writeLeaderInformation(LeaderInformation leaderInformation)
{
+ assert(running);
+ final UUID confirmedLeaderSessionID =
leaderInformation.getLeaderSessionID();
+ final String confirmedLeaderAddress =
leaderInformation.getLeaderAddress();
+ try {
+ kubeClient.checkAndUpdateConfigMap(
+ configMapName,
+ configMap -> {
+ if
(KubernetesLeaderElector.hasLeadership(configMap, lockIdentity)) {
+ // Get the updated ConfigMap
with new leader information
+ if (confirmedLeaderAddress ==
null) {
+
configMap.getData().remove(LEADER_ADDRESS_KEY);
+ } else {
+
configMap.getData().put(LEADER_ADDRESS_KEY, confirmedLeaderAddress);
+ }
+ if (confirmedLeaderSessionID ==
null) {
+
configMap.getData().remove(LEADER_SESSION_ID_KEY);
+ } else {
+
configMap.getData().put(LEADER_SESSION_ID_KEY,
confirmedLeaderSessionID.toString());
+ }
+
configMap.getLabels().putAll(configMapLabels);
+ return Optional.of(configMap);
+ }
+ return Optional.empty();
+ }).get();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Successfully wrote leader information:
Leader={}, session ID={}.",
+ confirmedLeaderAddress,
+ confirmedLeaderSessionID);
+ }
+ } catch (Exception e) {
+ fatalErrorHandler.onFatalError(
+ new KubernetesException("Could not write leader
information since ConfigMap " + configMapName
+ + " does not exist.", e));
+ }
+ }
+
+ @Override
+ public boolean hasLeadership() {
+ assert(running);
+ final Optional<KubernetesConfigMap> configMapOpt =
kubeClient.getConfigMap(configMapName);
+ if (configMapOpt.isPresent()) {
+ return
KubernetesLeaderElector.hasLeadership(configMapOpt.get(), lockIdentity);
+ } else {
+ fatalErrorHandler.onFatalError(
+ new KubernetesException("ConfigMap " +
configMapName + " does not exist.", null));
+ return false;
+ }
+ }
+
+ private class LeaderCallbackHandlerImpl extends
KubernetesLeaderElector.LeaderCallbackHandler {
+
+ @Override
+ public void isLeader() {
+ leaderElectionEventHandler.onGrantLeadership();
+ }
+
+ @Override
+ public void notLeader() {
+ leaderElectionEventHandler.onRevokeLeadership();
+ // Continue to contend the leader
+ leaderElector.run();
+ }
+ }
+
+ private class ConfigMapCallbackHandlerImpl implements
FlinkKubeClient.WatchCallbackHandler<KubernetesConfigMap> {
+ @Override
+ public void onAdded(List<KubernetesConfigMap> configMaps) {
+ // noop
+ }
+
+ @Override
+ public void onModified(List<KubernetesConfigMap> configMaps) {
+ // We should only receive events for the watched
ConfigMap
+ final KubernetesConfigMap configMap =
checkConfigMaps(configMaps, configMapName);
+
+ if (KubernetesLeaderElector.hasLeadership(configMap,
lockIdentity)) {
+
leaderElectionEventHandler.onLeaderInformationChange(getLeaderInformationFromConfigMap(configMap));
Review comment:
Copy what I have added in the description of
`LeaderElectionEventHandler#onLeaderInformationChange`. Does it make sense? Or
you still believe that we should keep a local cache in
`KubernetesLeaderElectionDriver` to check whether the leader information is
truly changed.
```
Duplicated leader change events could happen, so the implementation should
check whether the passed leader information is really different with internal
confirmed leader information.
```
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionDriverTest.java
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.runtime.leaderelection.LeaderInformation;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+
+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.apache.flink.kubernetes.utils.Constants.LEADER_ADDRESS_KEY;
+import static
org.apache.flink.kubernetes.utils.Constants.LEADER_SESSION_ID_KEY;
+import static org.hamcrest.Matchers.containsString;
+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 KubernetesLeaderElectionDriver}.
+ */
+public class KubernetesLeaderElectionDriverTest extends
KubernetesHighAvailabilityTestBase {
+
+ @Test
+ public void testIsLeader() throws Exception {
+ new Context() {{
+ runTest(
+ () -> {
+ // Grant leadership
+ leaderCallbackGrantLeadership();
+
assertThat(electionEventHandler.isLeader(), is(true));
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(LEADER_INFORMATION));
+ });
+ }};
+ }
+
+ @Test
+ public void testNotLeader() throws Exception {
+ new Context() {{
+ runTest(
+ () -> {
+ leaderCallbackGrantLeadership();
+ // Revoke leadership
+ getLeaderCallback().notLeader();
+
+
electionEventHandler.waitForRevokeLeader(TIMEOUT);
+
assertThat(electionEventHandler.isLeader(), is(false));
+
assertThat(electionEventHandler.getConfirmedLeaderInformation(),
is(LeaderInformation.empty()));
+ // The ConfigMap should also be cleared
+
assertThat(getLeaderConfigMap().getData().get(LEADER_ADDRESS_KEY),
is(nullValue()));
+
assertThat(getLeaderConfigMap().getData().get(LEADER_SESSION_ID_KEY),
is(nullValue()));
Review comment:
Because we do not have new leader elected and the lock identity
annotation on the ConfigMap is still the old one. So the
`KubernetesLeaderElector#hasLeadership` could pass the check. I think this is
the expected behavior. Once the leader ConfigMap is over written by a new
leader. It should return false.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.annotation.VisibleForTesting;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.Objects;
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation for leader election service. Composed with different
{@link LeaderElectionDriver}, we could
+ * perform a leader election for the contender, and then persist the leader
information to various storage.
+ */
+public class DefaultLeaderElectionService implements LeaderElectionService,
LeaderElectionEventHandler {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DefaultLeaderElectionService.class);
+
+ private final Object lock = new Object();
+
+ private final LeaderElectionDriverFactory leaderElectionDriverFactory;
+
+ /** The leader contender which applies for leadership. */
+ @GuardedBy("lock")
+ private volatile LeaderContender leaderContender;
+
+ @GuardedBy("lock")
+ private volatile UUID issuedLeaderSessionID;
+
+ @GuardedBy("lock")
+ private volatile UUID confirmedLeaderSessionID;
+
+ @GuardedBy("lock")
+ private volatile String confirmedLeaderAddress;
+
+ @GuardedBy("lock")
+ private volatile boolean running;
+
+ @GuardedBy("lock")
+ private LeaderElectionDriver leaderElectionDriver;
+
+ public DefaultLeaderElectionService(LeaderElectionDriverFactory
leaderElectionDriverFactory) {
+ this.leaderElectionDriverFactory =
checkNotNull(leaderElectionDriverFactory);
+
+ leaderContender = null;
+
+ issuedLeaderSessionID = null;
+ confirmedLeaderSessionID = null;
+ confirmedLeaderAddress = null;
+
+ running = false;
+ }
+
+ @Override
+ public final void start(LeaderContender contender) throws Exception {
+ checkNotNull(contender, "Contender must not be null.");
+ Preconditions.checkState(leaderContender == null, "Contender
was already set.");
+
+ synchronized (lock) {
+ leaderContender = contender;
+ leaderElectionDriver =
leaderElectionDriverFactory.createLeaderElectionDriver(
+ this, new LeaderElectionFatalErrorHandler(),
leaderContender.getDescription());
+ LOG.info("Starting DefaultLeaderElectionService with
{}.", leaderElectionDriver);
+
+ running = true;
+ }
+ }
+
+ @Override
+ public final void stop() throws Exception {
+ LOG.info("Stopping DefaultLeaderElectionService.");
+
+ synchronized (lock) {
+ if (!running) {
+ return;
+ }
+ running = false;
+ clearConfirmedLeaderInformation();
+ leaderElectionDriver.close();
+ }
+ }
+
+ @Override
+ public void confirmLeadership(UUID leaderSessionID, String
leaderAddress) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Confirm leader session ID {} for leader {}.",
+ leaderSessionID,
+ leaderAddress);
+ }
+
+ checkNotNull(leaderSessionID);
+
+ synchronized (lock) {
+ if (hasLeadership(leaderSessionID)) {
+ if (running) {
+
confirmLeaderInformation(leaderSessionID, leaderAddress);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring the leader
session Id {} confirmation, since the " +
+ "LeaderElectionService
has already been stopped.", leaderSessionID);
+ }
+ }
+ } else {
+ // Received an old confirmation call
+ if
(!leaderSessionID.equals(this.issuedLeaderSessionID)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Receive an old
confirmation call of leader session ID {}, " +
+ "current issued session
ID is {}", leaderSessionID, issuedLeaderSessionID);
+ }
+ } else {
+ LOG.warn("The leader session ID {} was
confirmed even though the " +
+ "corresponding JobManager was
not elected as the leader.", leaderSessionID);
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean hasLeadership(@Nonnull UUID leaderSessionId) {
+ synchronized (lock) {
+ if (running) {
+ return leaderElectionDriver.hasLeadership() &&
leaderSessionId.equals(issuedLeaderSessionID);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("hasLeadership is called
after the service is stopped, returning false.");
+ }
+ return false;
+ }
+ }
+ }
+
+ /**
+ * 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
+ */
+ @VisibleForTesting
+ @Nullable
+ public UUID getLeaderSessionID() {
+ return confirmedLeaderSessionID;
+ }
+
+ @GuardedBy("lock")
+ private void confirmLeaderInformation(UUID leaderSessionID, String
leaderAddress) {
+ confirmedLeaderSessionID = leaderSessionID;
+ confirmedLeaderAddress = leaderAddress;
+ leaderElectionDriver.writeLeaderInformation(
+ LeaderInformation.known(confirmedLeaderSessionID,
confirmedLeaderAddress));
+ }
+
+ @GuardedBy("lock")
+ private void clearConfirmedLeaderInformation() {
+ confirmedLeaderSessionID = null;
+ confirmedLeaderAddress = null;
+ }
+
+ @Override
+ @GuardedBy("lock")
+ public void onGrantLeadership() {
+ synchronized (lock) {
+ if (running) {
+ issuedLeaderSessionID = UUID.randomUUID();
+ clearConfirmedLeaderInformation();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Grant leadership to contender
{} with session ID {}.",
+
leaderContender.getDescription(),
+ issuedLeaderSessionID);
+ }
+
+
leaderContender.grantLeadership(issuedLeaderSessionID);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring the grant
leadership notification since the {} has " +
+ "already been closed.",
leaderElectionDriver);
+ }
+ }
+ }
+ }
+
+ @Override
+ @GuardedBy("lock")
+ public void onRevokeLeadership() {
+ synchronized (lock) {
+ if (running) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Revoke leadership of {}
({}@{}).",
+
leaderContender.getDescription(),
+ confirmedLeaderSessionID,
+ confirmedLeaderAddress);
+ }
+
+ issuedLeaderSessionID = null;
+ clearConfirmedLeaderInformation();
+
+ leaderContender.revokeLeadership();
+
+ // Clear the old leader information on the
external storage
+
leaderElectionDriver.writeLeaderInformation(LeaderInformation.empty());
Review comment:
The reason why I want to do this is to process the following situation.
* Old leader lost leadership and new leader is not launched very fast(maybe
120s or longer).
* Old leader should clear the leader information on ConfigMap so that the
listener could know this event.
For Kubernetes, it could work. Because no new leader has been elected and
the lock identity annotation is still the old one.
----------------------------------------------------------------
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:
us...@infra.apache.org
