You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2024/02/13 16:19:29 UTC
(flink) branch release-1.18 updated (d69393678ef -> 35c560312ef)
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a change to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
from d69393678ef [FLINK-34422][test] BatchTestBase uses MiniClusterExtension
new e5d488c9e5b [hotfix][test] Moves ConfigMap lifecycle management into @BeforeEach/@AfterEach methods
new 6441f5845f7 [hotfix][test] Makes ManuallyTriggeredScheduledExecutorService#execute rely on a BlockingQueue
new 38690d06e00 [hotfix][test] Refactors TestingLeaderCallbackHandler to allow async calls
new 0afa89d2840 [hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding tasks
new 87560b7cedd [FLINK-34333][k8s] fabric8io LeaderElector is created with every new #run() call (cherry-picked from FLINK-34007)
new 35c560312ef [FLINK-34333][k8s] Upgrade k8s client to v6.9.2 to cover client issue #5464 (cherry-picked from FLINK-34007)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../java/org/apache/flink/util/ExecutorUtils.java | 14 ++-
flink-kubernetes/pom.xml | 2 +-
.../kubeclient/FlinkKubeClientFactory.java | 35 +++---
.../resources/KubernetesLeaderElector.java | 77 ++++++++++--
.../src/main/resources/META-INF/NOTICE | 52 ++++----
.../flink/kubernetes/KubernetesClientTestBase.java | 2 +-
.../resources/KubernetesLeaderElectorITCase.java | 132 ++++++++++++++++++++-
.../resources/TestingLeaderCallbackHandler.java | 42 +++++--
.../ManuallyTriggeredScheduledExecutorService.java | 44 ++++---
9 files changed, 311 insertions(+), 89 deletions(-)
(flink) 04/06: [hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding tasks
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0afa89d2840b23b606329e1885c8d622c6960645
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Fri Feb 2 14:34:47 2024 +0100
[hotfix][core] Makes ExecutorUtils#gracefulShutdown return any outstanding tasks
---
.../src/main/java/org/apache/flink/util/ExecutorUtils.java | 14 ++++++++++----
1 file changed, 10 insertions(+), 4 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
index 939ffd6624c..99e2f3317e5 100644
--- a/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/util/ExecutorUtils.java
@@ -21,6 +21,8 @@ package org.apache.flink.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@@ -38,8 +40,9 @@ public class ExecutorUtils {
* @param timeout to wait for the termination of all ExecutorServices
* @param unit of the timeout
* @param executorServices to shut down
+ * @return Tasks that were not executed prior to a {@link ExecutorService#shutdownNow()}.
*/
- public static void gracefulShutdown(
+ public static List<Runnable> gracefulShutdown(
long timeout, TimeUnit unit, ExecutorService... executorServices) {
for (ExecutorService executorService : executorServices) {
executorService.shutdown();
@@ -50,22 +53,23 @@ public class ExecutorUtils {
long timeLeft = unit.toMillis(timeout);
boolean hasTimeLeft = timeLeft > 0L;
+ final List<Runnable> outstandingTasks = new ArrayList<>();
for (ExecutorService executorService : executorServices) {
if (wasInterrupted || !hasTimeLeft) {
- executorService.shutdownNow();
+ outstandingTasks.addAll(executorService.shutdownNow());
} else {
try {
if (!executorService.awaitTermination(timeLeft, TimeUnit.MILLISECONDS)) {
LOG.warn(
"ExecutorService did not terminate in time. Shutting it down now.");
- executorService.shutdownNow();
+ outstandingTasks.addAll(executorService.shutdownNow());
}
} catch (InterruptedException e) {
LOG.warn(
"Interrupted while shutting down executor services. Shutting all "
+ "remaining ExecutorServices down now.",
e);
- executorService.shutdownNow();
+ outstandingTasks.addAll(executorService.shutdownNow());
wasInterrupted = true;
@@ -76,6 +80,8 @@ public class ExecutorUtils {
hasTimeLeft = timeLeft > 0L;
}
}
+
+ return outstandingTasks;
}
/**
(flink) 05/06: [FLINK-34333][k8s] fabric8io LeaderElector is created with every new #run() call (cherry-picked from FLINK-34007)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 87560b7cedd6c857612a24b83485f5000b9edbd6
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Tue Jan 23 16:27:02 2024 +0100
[FLINK-34333][k8s] fabric8io LeaderElector is created with every new #run() call (cherry-picked from FLINK-34007)
v5.12.4 allowed us to reuse the LeaderElector. With v6.6.2 (fabric8io/kubernetes-client#4125) this behavior changed. One LeaderElector can only be used until the leadership is lost.
An ITCase is added to cover the scenario where the leadership is lost.
---
.../kubeclient/FlinkKubeClientFactory.java | 35 ++++---
.../resources/KubernetesLeaderElector.java | 77 ++++++++++++---
.../resources/KubernetesLeaderElectorITCase.java | 108 +++++++++++++++++++++
3 files changed, 193 insertions(+), 27 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClientFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClientFactory.java
index 99a9ffc9a33..c16508f8b99 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClientFactory.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClientFactory.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.kubeclient;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.util.FileUtils;
@@ -46,16 +47,8 @@ public class FlinkKubeClientFactory {
return INSTANCE;
}
- /**
- * Create a Flink Kubernetes client with the given configuration.
- *
- * @param flinkConfig Flink configuration
- * @param useCase Flink Kubernetes client use case (e.g. client, resourcemanager,
- * kubernetes-ha-services)
- * @return Return the Flink Kubernetes client with the specified configuration and dedicated IO
- * executor.
- */
- public FlinkKubeClient fromConfiguration(Configuration flinkConfig, String useCase) {
+ @VisibleForTesting
+ public NamespacedKubernetesClient createFabric8ioKubernetesClient(Configuration flinkConfig) {
final Config config;
final String kubeContext = flinkConfig.getString(KubernetesConfigOptions.CONTEXT);
@@ -95,11 +88,23 @@ public class FlinkKubeClientFactory {
config.setUserAgent(userAgent);
LOG.debug("Setting Kubernetes client namespace: {}, userAgent: {}", namespace, userAgent);
- final NamespacedKubernetesClient client =
- new KubernetesClientBuilder()
- .withConfig(config)
- .build()
- .adapt(NamespacedKubernetesClient.class);
+ return new KubernetesClientBuilder()
+ .withConfig(config)
+ .build()
+ .adapt(NamespacedKubernetesClient.class);
+ }
+
+ /**
+ * Create a Flink Kubernetes client with the given configuration.
+ *
+ * @param flinkConfig Flink configuration
+ * @param useCase Flink Kubernetes client use case (e.g. client, resourcemanager,
+ * kubernetes-ha-services)
+ * @return Return the Flink Kubernetes client with the specified configuration and dedicated IO
+ * executor.
+ */
+ public FlinkKubeClient fromConfiguration(Configuration flinkConfig, String useCase) {
+ final NamespacedKubernetesClient client = createFabric8ioKubernetesClient(flinkConfig);
final int poolSize =
flinkConfig.get(KubernetesConfigOptions.KUBERNETES_CLIENT_IO_EXECUTOR_POOL_SIZE);
return new Fabric8FlinkKubeClient(
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
index d072e15cf18..dd0accf4369 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElector.java
@@ -20,7 +20,9 @@ package org.apache.flink.kubernetes.kubeclient.resources;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
+import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.FutureUtils;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
@@ -31,8 +33,13 @@ import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.ConfigM
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
/**
* Represent {@link KubernetesLeaderElector} in kubernetes. {@link LeaderElector#run()} is a
@@ -55,17 +62,32 @@ public class KubernetesLeaderElector {
private final Object lock = new Object();
- private final ExecutorService executorService =
- Executors.newFixedThreadPool(
- 3, new ExecutorThreadFactory("KubernetesLeaderElector-ExecutorService"));
+ private final NamespacedKubernetesClient kubernetesClient;
+ private final LeaderElectionConfig leaderElectionConfig;
+ private final ExecutorService executorService;
- private final LeaderElector internalLeaderElector;
+ private CompletableFuture<?> currentLeaderElectionSession = FutureUtils.completedVoidFuture();
public KubernetesLeaderElector(
NamespacedKubernetesClient kubernetesClient,
KubernetesLeaderElectionConfiguration leaderConfig,
LeaderCallbackHandler leaderCallbackHandler) {
- final LeaderElectionConfig leaderElectionConfig =
+ this(
+ kubernetesClient,
+ leaderConfig,
+ leaderCallbackHandler,
+ Executors.newSingleThreadExecutor(
+ new ExecutorThreadFactory("KubernetesLeaderElector-ExecutorService")));
+ }
+
+ @VisibleForTesting
+ public KubernetesLeaderElector(
+ NamespacedKubernetesClient kubernetesClient,
+ KubernetesLeaderElectionConfiguration leaderConfig,
+ LeaderCallbackHandler leaderCallbackHandler,
+ ExecutorService executorService) {
+ this.kubernetesClient = kubernetesClient;
+ this.leaderElectionConfig =
new LeaderElectionConfigBuilder()
.withName(leaderConfig.getConfigMapName())
.withLeaseDuration(leaderConfig.getLeaseDuration())
@@ -76,6 +98,7 @@ public class KubernetesLeaderElector {
leaderConfig.getLockIdentity()))
.withRenewDeadline(leaderConfig.getRenewDeadline())
.withRetryPeriod(leaderConfig.getRetryPeriod())
+ .withReleaseOnCancel(true)
.withLeaderCallbacks(
new LeaderCallbacks(
leaderCallbackHandler::isLeader,
@@ -86,12 +109,27 @@ public class KubernetesLeaderElector {
newLeader,
leaderConfig.getConfigMapName())))
.build();
- internalLeaderElector =
- new LeaderElector(kubernetesClient, leaderElectionConfig, executorService);
+ this.executorService = executorService;
+
LOG.info(
- "Create KubernetesLeaderElector {} with lock identity {}.",
- leaderConfig.getConfigMapName(),
- leaderConfig.getLockIdentity());
+ "Create KubernetesLeaderElector on lock {}.",
+ leaderElectionConfig.getLock().describe());
+ }
+
+ @GuardedBy("lock")
+ private void resetInternalLeaderElector() {
+ cancelCurrentLeaderElectionSession();
+
+ currentLeaderElectionSession =
+ new LeaderElector(kubernetesClient, leaderElectionConfig, executorService).start();
+
+ LOG.info(
+ "Triggered leader election on lock {}.", leaderElectionConfig.getLock().describe());
+ }
+
+ @GuardedBy("lock")
+ private void cancelCurrentLeaderElectionSession() {
+ currentLeaderElectionSession.cancel(true);
}
public void run() {
@@ -100,14 +138,29 @@ public class KubernetesLeaderElector {
LOG.debug(
"Ignoring KubernetesLeaderElector.run call because the leader elector has already been shut down.");
} else {
- executorService.execute(internalLeaderElector::run);
+ resetInternalLeaderElector();
}
}
}
public void stop() {
synchronized (lock) {
- executorService.shutdownNow();
+ // cancelling the current session needs to happen explicitly to allow the execution of
+ // code that handles the leader loss
+ cancelCurrentLeaderElectionSession();
+
+ // the shutdown of the executor needs to happen gracefully for scenarios where the
+ // release is called in the executorService. Interrupting this logic will result in the
+ // leadership-lost event not being sent to the client.
+ final List<Runnable> outStandingTasks =
+ ExecutorUtils.gracefulShutdown(30, TimeUnit.SECONDS, executorService);
+
+ if (!outStandingTasks.isEmpty()) {
+ LOG.warn(
+ "{} events were not processed before stopping the {} instance.",
+ outStandingTasks.size(),
+ KubernetesLeaderElector.class.getSimpleName());
+ }
}
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
index a86008f693a..e167d058765 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
@@ -19,18 +19,23 @@
package org.apache.flink.kubernetes.kubeclient.resources;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.ManuallyTriggeredScheduledExecutorService;
import org.apache.flink.kubernetes.KubernetesExtension;
+import org.apache.flink.kubernetes.configuration.KubernetesHighAvailabilityOptions;
import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfiguration;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClientFactory;
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.time.Duration;
import java.util.UUID;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -124,4 +129,107 @@ class KubernetesLeaderElectorITCase {
}
}
}
+
+ /**
+ * This test verifies that the {@link KubernetesLeaderElector} is able to handle scenario where
+ * the lease cannot be renewed.
+ *
+ * <p>See FLINK-34007 for further details.
+ */
+ @Test
+ void testLeaderElectorLifecycleManagement() throws Exception {
+ final Configuration configuration = kubernetesExtension.getConfiguration();
+
+ try (final NamespacedKubernetesClient client =
+ kubeClientFactory.createFabric8ioKubernetesClient(configuration)) {
+
+ // set a low timeout that makes the client stop renewing the leadership lease
+ final Duration renewTimeout = Duration.ofMillis(100);
+ configuration.set(
+ KubernetesHighAvailabilityOptions.KUBERNETES_RENEW_DEADLINE, renewTimeout);
+
+ final String lockIdentity = UUID.randomUUID().toString();
+ final KubernetesLeaderElectionConfiguration leaderConfig =
+ new KubernetesLeaderElectionConfiguration(
+ configMapName, lockIdentity, configuration);
+ final TestingLeaderCallbackHandler leadershipCallbackHandler =
+ new TestingLeaderCallbackHandler(lockIdentity);
+
+ final ManuallyTriggeredScheduledExecutorService executorService =
+ new ManuallyTriggeredScheduledExecutorService();
+ final KubernetesLeaderElector testInstance =
+ new KubernetesLeaderElector(
+ client, leaderConfig, leadershipCallbackHandler, executorService);
+
+ // first leadership lifecycle initiation
+ testInstance.run();
+
+ // triggers acquiring the leadership
+ final Duration waitForNextTaskForever = Duration.ofDays(10);
+ executorService.trigger(waitForNextTaskForever);
+
+ assertThatFuture(leadershipCallbackHandler.waitForNewLeaderAsync())
+ .as("The leadership should be acquired eventually.")
+ .eventuallySucceeds();
+
+ // halt thread to reach the renew deadline
+ Thread.sleep(renewTimeout.plusSeconds(1).toMillis());
+
+ // triggers renew loop within fabric8io's LeaderElector
+ executorService.trigger();
+
+ assertThatFuture(leadershipCallbackHandler.waitForRevokeLeaderAsync())
+ .as(
+ "The leadership should be lost eventually due to the renewal loop being stopped.")
+ .eventuallySucceeds();
+
+ // revoking the leadership initiates another leadership lifecycle
+ testInstance.run();
+ executorService.trigger(waitForNextTaskForever);
+
+ assertThatFuture(leadershipCallbackHandler.waitForNewLeaderAsync())
+ .as("The leadership should be acquired eventually again.");
+ }
+ }
+
+ @Test
+ void testKubernetesLeaderElectorSendingLeadershipLostSignalOnStop() {
+ final Configuration configuration = kubernetesExtension.getConfiguration();
+
+ try (final NamespacedKubernetesClient client =
+ kubeClientFactory.createFabric8ioKubernetesClient(configuration)) {
+
+ final String lockIdentity = UUID.randomUUID().toString();
+ final KubernetesLeaderElectionConfiguration leaderConfig =
+ new KubernetesLeaderElectionConfiguration(
+ configMapName, lockIdentity, configuration);
+ final TestingLeaderCallbackHandler leadershipCallbackHandler =
+ new TestingLeaderCallbackHandler(lockIdentity);
+
+ final ManuallyTriggeredScheduledExecutorService executorService =
+ new ManuallyTriggeredScheduledExecutorService();
+ final KubernetesLeaderElector testInstance =
+ new KubernetesLeaderElector(
+ client, leaderConfig, leadershipCallbackHandler, executorService);
+
+ // initiate leadership lifecycle
+ testInstance.run();
+
+ final Duration waitForNextTaskForever = Duration.ofDays(10);
+ executorService.trigger(waitForNextTaskForever);
+ assertThatFuture(leadershipCallbackHandler.waitForNewLeaderAsync())
+ .as("Leadership should be acquired eventually.")
+ .eventuallySucceeds();
+
+ testInstance.stop();
+
+ assertThat(leadershipCallbackHandler.hasLeadership())
+ .as("Leadership should be lost right away after stopping the test instance.")
+ .isFalse();
+
+ assertThatFuture(leadershipCallbackHandler.waitForRevokeLeaderAsync())
+ .as("There should be a leadership lost event being received eventually.")
+ .eventuallySucceeds();
+ }
+ }
}
(flink) 01/06: [hotfix][test] Moves ConfigMap lifecycle management into @BeforeEach/@AfterEach methods
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e5d488c9e5b12b1bd16ceeb4f44fe0526777b802
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Fri Jan 26 11:25:34 2024 +0100
[hotfix][test] Moves ConfigMap lifecycle management into @BeforeEach/@AfterEach methods
---
.../resources/KubernetesLeaderElectorITCase.java | 24 +++++++++++++++++-----
1 file changed, 19 insertions(+), 5 deletions(-)
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
index e97d23c6d54..a86008f693a 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesLeaderElectorITCase.java
@@ -24,6 +24,8 @@ import org.apache.flink.kubernetes.configuration.KubernetesLeaderElectionConfigu
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClientFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -41,14 +43,27 @@ class KubernetesLeaderElectorITCase {
private final FlinkKubeClientFactory kubeClientFactory = new FlinkKubeClientFactory();
- private static final String LEADER_CONFIGMAP_NAME_PREFIX = "leader-test-cluster";
+ private String configMapName;
+
+ @BeforeEach
+ void initializeConfigMapName() {
+ this.configMapName =
+ String.format(
+ "%s-configmap-%s",
+ // needs to comply to RFC-1123
+ KubernetesLeaderElectorITCase.class.getSimpleName().toLowerCase(),
+ UUID.randomUUID());
+ }
+
+ @AfterEach
+ void deleteConfigMapName() {
+ kubernetesExtension.getFlinkKubeClient().deleteConfigMap(this.configMapName).join();
+ }
@Test
void testMultipleKubernetesLeaderElectors() throws Exception {
final Configuration configuration = kubernetesExtension.getConfiguration();
- final String leaderConfigMapName =
- LEADER_CONFIGMAP_NAME_PREFIX + System.currentTimeMillis();
final int leaderNum = 3;
final KubernetesLeaderElector[] leaderElectors = new KubernetesLeaderElector[leaderNum];
@@ -64,7 +79,7 @@ class KubernetesLeaderElectorITCase {
new TestingLeaderCallbackHandler(UUID.randomUUID().toString());
final KubernetesLeaderElectionConfiguration leaderConfig =
new KubernetesLeaderElectionConfiguration(
- leaderConfigMapName,
+ configMapName,
leaderCallbackHandlers[i].getLockIdentity(),
configuration);
leaderElectors[i] =
@@ -107,7 +122,6 @@ class KubernetesLeaderElectorITCase {
kubeClients[i].close();
}
}
- kubernetesExtension.getFlinkKubeClient().deleteConfigMap(leaderConfigMapName).get();
}
}
}
(flink) 06/06: [FLINK-34333][k8s] Upgrade k8s client to v6.9.2 to cover client issue #5464 (cherry-picked from FLINK-34007)
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 35c560312efc91dafd1b4674ce1e10acc9320ab1
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Mon Jan 29 11:29:07 2024 +0100
[FLINK-34333][k8s] Upgrade k8s client to v6.9.2 to cover client issue #5464 (cherry-picked from FLINK-34007)
https://github.com/fabric8io/kubernetes-client/issues/5463
The change in KubernetesClientTestBase was necessary because of the change [#5369](https://github.com/fabric8io/kubernetes-client/pull/5369) that ended up in v6.9.0 as well.
---
flink-kubernetes/pom.xml | 2 +-
.../src/main/resources/META-INF/NOTICE | 52 +++++++++++-----------
.../flink/kubernetes/KubernetesClientTestBase.java | 2 +-
3 files changed, 28 insertions(+), 28 deletions(-)
diff --git a/flink-kubernetes/pom.xml b/flink-kubernetes/pom.xml
index 7faf4a71f8b..e1a0af59e63 100644
--- a/flink-kubernetes/pom.xml
+++ b/flink-kubernetes/pom.xml
@@ -31,7 +31,7 @@ under the License.
<packaging>jar</packaging>
<properties>
- <kubernetes.client.version>6.6.2</kubernetes.client.version>
+ <kubernetes.client.version>6.9.2</kubernetes.client.version>
<surefire.module.config><!--
CommonTestUtils#setEnv
-->--add-opens=java.base/java.util=ALL-UNNAMED
diff --git a/flink-kubernetes/src/main/resources/META-INF/NOTICE b/flink-kubernetes/src/main/resources/META-INF/NOTICE
index f295c017ff4..2cc796fe022 100644
--- a/flink-kubernetes/src/main/resources/META-INF/NOTICE
+++ b/flink-kubernetes/src/main/resources/META-INF/NOTICE
@@ -14,31 +14,31 @@ This project bundles the following dependencies under the Apache Software Licens
- com.squareup.okhttp3:logging-interceptor:3.14.9
- com.squareup.okhttp3:okhttp:3.14.9
- com.squareup.okio:okio:1.17.2
-- io.fabric8:kubernetes-client:6.6.2
-- io.fabric8:kubernetes-client-api:6.6.2
-- io.fabric8:kubernetes-httpclient-okhttp:6.6.2
-- io.fabric8:kubernetes-model-admissionregistration:6.6.2
-- io.fabric8:kubernetes-model-apiextensions:6.6.2
-- io.fabric8:kubernetes-model-apps:6.6.2
-- io.fabric8:kubernetes-model-autoscaling:6.6.2
-- io.fabric8:kubernetes-model-batch:6.6.2
-- io.fabric8:kubernetes-model-certificates:6.6.2
-- io.fabric8:kubernetes-model-common:6.6.2
-- io.fabric8:kubernetes-model-coordination:6.6.2
-- io.fabric8:kubernetes-model-core:6.6.2
-- io.fabric8:kubernetes-model-discovery:6.6.2
-- io.fabric8:kubernetes-model-events:6.6.2
-- io.fabric8:kubernetes-model-extensions:6.6.2
-- io.fabric8:kubernetes-model-flowcontrol:6.6.2
-- io.fabric8:kubernetes-model-gatewayapi:6.6.2
-- io.fabric8:kubernetes-model-metrics:6.6.2
-- io.fabric8:kubernetes-model-networking:6.6.2
-- io.fabric8:kubernetes-model-node:6.6.2
-- io.fabric8:kubernetes-model-policy:6.6.2
-- io.fabric8:kubernetes-model-rbac:6.6.2
-- io.fabric8:kubernetes-model-resource:6.6.2
-- io.fabric8:kubernetes-model-scheduling:6.6.2
-- io.fabric8:kubernetes-model-storageclass:6.6.2
+- io.fabric8:kubernetes-client:6.9.2
+- io.fabric8:kubernetes-client-api:6.9.2
+- io.fabric8:kubernetes-httpclient-okhttp:6.9.2
+- io.fabric8:kubernetes-model-admissionregistration:6.9.2
+- io.fabric8:kubernetes-model-apiextensions:6.9.2
+- io.fabric8:kubernetes-model-apps:6.9.2
+- io.fabric8:kubernetes-model-autoscaling:6.9.2
+- io.fabric8:kubernetes-model-batch:6.9.2
+- io.fabric8:kubernetes-model-certificates:6.9.2
+- io.fabric8:kubernetes-model-common:6.9.2
+- io.fabric8:kubernetes-model-coordination:6.9.2
+- io.fabric8:kubernetes-model-core:6.9.2
+- io.fabric8:kubernetes-model-discovery:6.9.2
+- io.fabric8:kubernetes-model-events:6.9.2
+- io.fabric8:kubernetes-model-extensions:6.9.2
+- io.fabric8:kubernetes-model-flowcontrol:6.9.2
+- io.fabric8:kubernetes-model-gatewayapi:6.9.2
+- io.fabric8:kubernetes-model-metrics:6.9.2
+- io.fabric8:kubernetes-model-networking:6.9.2
+- io.fabric8:kubernetes-model-node:6.9.2
+- io.fabric8:kubernetes-model-policy:6.9.2
+- io.fabric8:kubernetes-model-rbac:6.9.2
+- io.fabric8:kubernetes-model-resource:6.9.2
+- io.fabric8:kubernetes-model-scheduling:6.9.2
+- io.fabric8:kubernetes-model-storageclass:6.9.2
- io.fabric8:zjsonpatch:0.3.0
-- org.snakeyaml:snakeyaml-engine:2.6
+- org.snakeyaml:snakeyaml-engine:2.7
- org.yaml:snakeyaml:1.33
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java
index 8d9a4af663b..aef667291b8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java
@@ -163,7 +163,7 @@ public class KubernetesClientTestBase extends KubernetesTestBase {
// mock four kinds of events.
String mockPath =
String.format(
- "/api/v1/namespaces/%s/pods?labelSelector=%s&resourceVersion=%s&allowWatchBookmarks=true&watch=true",
+ "/api/v1/namespaces/%s/pods?allowWatchBookmarks=true&labelSelector=%s&resourceVersion=%s&watch=true",
namespace,
labels.entrySet().stream()
.map(entry -> entry.getKey() + "%3D" + entry.getValue())
(flink) 02/06: [hotfix][test] Makes ManuallyTriggeredScheduledExecutorService#execute rely on a BlockingQueue
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6441f5845f75dad860c65d71aaca252973dbaba2
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Tue Jan 23 16:21:45 2024 +0100
[hotfix][test] Makes ManuallyTriggeredScheduledExecutorService#execute rely on a BlockingQueue
This allows us to wait for tasks to "arrive".
---
.../ManuallyTriggeredScheduledExecutorService.java | 44 ++++++++++++++--------
1 file changed, 29 insertions(+), 15 deletions(-)
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/ManuallyTriggeredScheduledExecutorService.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/ManuallyTriggeredScheduledExecutorService.java
index 0ae1d282c1b..7bd4c9dc74b 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/ManuallyTriggeredScheduledExecutorService.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/ManuallyTriggeredScheduledExecutorService.java
@@ -20,16 +20,18 @@ package org.apache.flink.core.testutils;
import javax.annotation.Nonnull;
-import java.util.ArrayDeque;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -44,7 +46,7 @@ import java.util.stream.Collectors;
*/
public class ManuallyTriggeredScheduledExecutorService implements ScheduledExecutorService {
- private final ArrayDeque<Runnable> queuedRunnables = new ArrayDeque<>();
+ private final BlockingQueue<Runnable> queuedRunnables = new LinkedBlockingQueue<>();
private final ConcurrentLinkedQueue<ScheduledTask<?>> nonPeriodicScheduledTasks =
new ConcurrentLinkedQueue<>();
@@ -60,9 +62,7 @@ public class ManuallyTriggeredScheduledExecutorService implements ScheduledExecu
@Override
public void execute(@Nonnull Runnable command) {
- synchronized (queuedRunnables) {
- queuedRunnables.addLast(command);
- }
+ queuedRunnables.add(command);
}
@Override
@@ -182,24 +182,38 @@ public class ManuallyTriggeredScheduledExecutorService implements ScheduledExecu
}
/**
- * Triggers the next queued runnable and executes it synchronously. This method throws an
- * exception if no Runnable is currently queued.
+ * Triggers the next task that was submitted for execution. The method blocks the given amount
+ * of time if no task is scheduled, yet.
+ *
+ * @param timeout The time to wait for a new task to be scheduled.
+ * @throws IllegalStateException if no task was scheduled in the given amount of time.
*/
- public void trigger() {
- final Runnable next;
+ public void trigger(Duration timeout) {
+ try {
+ final Runnable task = queuedRunnables.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
- synchronized (queuedRunnables) {
- next = queuedRunnables.removeFirst();
+ if (task == null) {
+ throw new IllegalStateException("No task was scheduled.");
+ }
+
+ task.run();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
}
+ }
- next.run();
+ /**
+ * Triggers the next task that was submitted for execution.
+ *
+ * @throws IllegalStateException if no task was scheduled before calling this method.
+ */
+ public void trigger() {
+ trigger(Duration.ZERO);
}
/** Gets the number of Runnables currently queued. */
public int numQueuedRunnables() {
- synchronized (queuedRunnables) {
- return queuedRunnables.size();
- }
+ return queuedRunnables.size();
}
public Collection<ScheduledFuture<?>> getActiveScheduledTasks() {
(flink) 03/06: [hotfix][test] Refactors TestingLeaderCallbackHandler to allow async calls
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 38690d06e00bb17fc13709ec1b7582c896ccb034
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Mon Jan 29 09:28:27 2024 +0100
[hotfix][test] Refactors TestingLeaderCallbackHandler to allow async calls
This way, we can use FlinkAssertions#assertThatFuture and use assertion messages instead of comments.
---
.../resources/TestingLeaderCallbackHandler.java | 42 ++++++++++++++++------
1 file changed, 32 insertions(+), 10 deletions(-)
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java
index bdfd86ff534..997128efd55 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/TestingLeaderCallbackHandler.java
@@ -18,9 +18,11 @@
package org.apache.flink.kubernetes.kubeclient.resources;
-import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.util.Preconditions;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.LinkedBlockingQueue;
/** Testing implementation for {@link KubernetesLeaderElector.LeaderCallbackHandler}. */
@@ -60,22 +62,42 @@ public class TestingLeaderCallbackHandler extends KubernetesLeaderElector.Leader
}
public static String waitUntilNewLeaderAppears() throws Exception {
- return sharedQueue.take();
+ return retrieveNextEventAsync(sharedQueue).get();
+ }
+
+ public static CompletableFuture<String> retrieveNextEventAsync(
+ BlockingQueue<String> eventQueue) {
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return eventQueue.take();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ }
+ });
}
public void waitForNewLeader() throws Exception {
- poll(leaderQueue);
+ waitForNewLeaderAsync().get();
+ }
+
+ public CompletableFuture<Void> waitForNewLeaderAsync() {
+ return waitForNextEvent(leaderQueue);
}
public void waitForRevokeLeader() throws Exception {
- poll(revokeQueue);
+ waitForRevokeLeaderAsync().get();
}
- private void poll(BlockingQueue<String> queue) throws Exception {
- CommonTestUtils.waitUntilCondition(
- () -> {
- final String lockIdentity = queue.take();
- return this.lockIdentity.equals(lockIdentity);
- });
+ public CompletableFuture<Void> waitForRevokeLeaderAsync() {
+ return waitForNextEvent(revokeQueue);
+ }
+
+ private CompletableFuture<Void> waitForNextEvent(BlockingQueue<String> eventQueue) {
+ return retrieveNextEventAsync(eventQueue)
+ .thenAccept(
+ eventLockIdentity ->
+ Preconditions.checkState(eventLockIdentity.equals(lockIdentity)));
}
}