You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2023/11/24 06:52:29 UTC
(flink) branch release-1.17 updated (cef22667ae6 -> 18d5a4696ec)
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from cef22667ae6 [FLINK-33582] Upgrade to flink-shaded 16.2
new 2a83c910eee [FLINK-31385] Introduce extended Assertj matchers for completable futures, that don't rely on timeouts.
new 18d5a4696ec [FLINK-33598][k8s] Watch HA configmap via name instead of lables to reduce pressure on APIserver
The 2 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:
...sMultipleComponentLeaderElectionHaServices.java | 11 +-
.../kubeclient/Fabric8FlinkKubeClient.java | 15 +--
.../kubernetes/kubeclient/FlinkKubeClient.java | 17 +--
.../KubernetesConfigMapSharedInformer.java | 17 ++-
.../KubernetesHighAvailabilityTestBase.java | 4 -
...KubernetesLeaderElectionAndRetrievalITCase.java | 5 +-
.../highavailability/KubernetesTestFixture.java | 31 +----
.../kubeclient/Fabric8FlinkKubeClientTest.java | 15 ---
.../kubeclient/TestingFlinkKubeClient.java | 34 ++----
.../resources/KubernetesSharedInformerITCase.java | 94 ++++----------
.../flink/core/testutils/FlinkAssertions.java | 27 ++++
.../testutils/FlinkCompletableFutureAssert.java | 136 +++++++++++++++++++++
.../flink/core/testutils/FlinkAssertionsTest.java | 121 ++++++++++++++++++
13 files changed, 336 insertions(+), 191 deletions(-)
create mode 100644 flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java
create mode 100644 flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java
(flink) 02/02: [FLINK-33598][k8s] Watch HA configmap via name instead of lables to reduce pressure on APIserver
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 18d5a4696eccac3b5e7fe1d579547feef4537c08
Author: Yun Tang <my...@live.com>
AuthorDate: Fri Nov 24 10:53:35 2023 +0800
[FLINK-33598][k8s] Watch HA configmap via name instead of lables to reduce pressure on APIserver
---
...sMultipleComponentLeaderElectionHaServices.java | 11 +--
.../kubeclient/Fabric8FlinkKubeClient.java | 15 +---
.../kubernetes/kubeclient/FlinkKubeClient.java | 17 +---
.../KubernetesConfigMapSharedInformer.java | 17 ++--
.../KubernetesHighAvailabilityTestBase.java | 4 -
...KubernetesLeaderElectionAndRetrievalITCase.java | 5 +-
.../highavailability/KubernetesTestFixture.java | 31 +------
.../kubeclient/Fabric8FlinkKubeClientTest.java | 15 ----
.../kubeclient/TestingFlinkKubeClient.java | 34 ++------
.../resources/KubernetesSharedInformerITCase.java | 94 ++++++----------------
10 files changed, 52 insertions(+), 191 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java
index 5af3cbb20e5..5e741561589 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionHaServices.java
@@ -52,7 +52,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
import static org.apache.flink.kubernetes.utils.Constants.NAME_SEPARATOR;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -95,9 +94,7 @@ public class KubernetesMultipleComponentLeaderElectionHaServices extends Abstrac
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.configMapSharedWatcher =
- this.kubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
+ this.kubeClient.createConfigMapSharedWatcher(getClusterConfigMap());
this.watchExecutorService =
Executors.newCachedThreadPool(
new ExecutorThreadFactory("config-map-watch-handler"));
@@ -216,11 +213,7 @@ public class KubernetesMultipleComponentLeaderElectionHaServices extends Abstrac
exception = e;
}
- kubeClient
- .deleteConfigMapsByLabels(
- KubernetesUtils.getConfigMapLabels(
- clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY))
- .get();
+ kubeClient.deleteConfigMap(getClusterConfigMap()).get();
ExceptionUtils.tryRethrowException(exception);
}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
index 023ed5df3d8..982dcbbd46d 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
@@ -332,15 +332,6 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient {
kubeClientExecutorService);
}
- @Override
- public CompletableFuture<Void> deleteConfigMapsByLabels(Map<String, String> labels) {
- // the only time, the delete method returns false is due to a 404 HTTP status which is
- // returned if the underlying resource doesn't exist
- return CompletableFuture.runAsync(
- () -> this.internalClient.configMaps().withLabels(labels).delete(),
- kubeClientExecutorService);
- }
-
@Override
public CompletableFuture<Void> deleteConfigMap(String configMapName) {
// the only time, the delete method returns false is due to a 404 HTTP status which is
@@ -351,9 +342,9 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient {
}
@Override
- public KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(
- Map<String, String> labels) {
- return new KubernetesConfigMapSharedInformer(this.internalClient, labels);
+ public KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(String name) {
+ LOG.info("Creating configmap shared watcher for {}.", name);
+ return new KubernetesConfigMapSharedInformer(this.internalClient, name);
}
@Override
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
index 64b45385b05..18139923b4e 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/FlinkKubeClient.java
@@ -164,16 +164,6 @@ public interface FlinkKubeClient extends AutoCloseable {
String configMapName,
Function<KubernetesConfigMap, Optional<KubernetesConfigMap>> updateFunction);
- /**
- * Delete the Kubernetes ConfigMaps by labels. This will be used by the HA service to clean up
- * all data.
- *
- * @param labels labels to filter the resources. e.g. type: high-availability
- * @return Return the delete future that only completes successfully, if the resources that are
- * subject to deletion are actually gone.
- */
- CompletableFuture<Void> deleteConfigMapsByLabels(Map<String, String> labels);
-
/**
* Delete a Kubernetes ConfigMap by name.
*
@@ -183,13 +173,12 @@ public interface FlinkKubeClient extends AutoCloseable {
CompletableFuture<Void> deleteConfigMap(String configMapName);
/**
- * Create a shared watcher for ConfigMaps with specified labels.
+ * Create a shared watcher for ConfigMaps with specified name.
*
- * @param labels labels to filter ConfigMaps. If the labels is null or empty, all ConfigMaps
- * will be taken in account, or it will be rejected if the implementation does not allow it.
+ * @param name name of the ConfigMap to watch.
* @return Return a shared watcher.
*/
- KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(Map<String, String> labels);
+ KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(String name);
/** Close the Kubernetes client with no exception. */
void close();
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesConfigMapSharedInformer.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesConfigMapSharedInformer.java
index a155bad60b8..057f3f11947 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesConfigMapSharedInformer.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesConfigMapSharedInformer.java
@@ -19,29 +19,26 @@
package org.apache.flink.kubernetes.kubeclient.resources;
import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
-import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import io.fabric8.kubernetes.client.dsl.Informable;
-import java.util.Map;
-
/** The shared informer for {@link ConfigMap}, it can be used as a shared watcher. */
public class KubernetesConfigMapSharedInformer
extends KubernetesSharedInformer<ConfigMap, KubernetesConfigMap>
implements KubernetesConfigMapSharedWatcher {
- public KubernetesConfigMapSharedInformer(
- NamespacedKubernetesClient client, Map<String, String> labels) {
- super(client, getInformableConfigMaps(client, labels), KubernetesConfigMap::new);
+ public KubernetesConfigMapSharedInformer(NamespacedKubernetesClient client, String name) {
+ super(client, getInformabaleConfigMaps(client, name), KubernetesConfigMap::new);
}
- private static Informable<ConfigMap> getInformableConfigMaps(
- NamespacedKubernetesClient client, Map<String, String> labels) {
+ private static Informable<ConfigMap> getInformabaleConfigMaps(
+ NamespacedKubernetesClient client, String name) {
Preconditions.checkArgument(
- !CollectionUtil.isNullOrEmpty(labels), "Labels must not be null or empty");
- return client.configMaps().withLabels(labels);
+ !StringUtils.isNullOrWhitespaceOnly(name), "Name must not be null or empty");
+ return client.configMaps().withName(name);
}
}
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHighAvailabilityTestBase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHighAvailabilityTestBase.java
index b7daa51a5bd..630769f5002 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHighAvailabilityTestBase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesHighAvailabilityTestBase.java
@@ -35,7 +35,6 @@ import org.apache.flink.util.function.RunnableWithException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
-import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -83,7 +82,6 @@ class KubernetesHighAvailabilityTestBase {
final Configuration configuration;
final CompletableFuture<Void> closeKubeClientFuture;
- final CompletableFuture<Map<String, String>> deleteConfigMapByLabelsFuture;
Context() {
kubernetesTestFixture =
@@ -91,8 +89,6 @@ class KubernetesHighAvailabilityTestBase {
flinkKubeClient = kubernetesTestFixture.getFlinkKubeClient();
configuration = kubernetesTestFixture.getConfiguration();
closeKubeClientFuture = kubernetesTestFixture.getCloseKubeClientFuture();
- deleteConfigMapByLabelsFuture =
- kubernetesTestFixture.getDeleteConfigMapByLabelsFuture();
electionEventHandler = new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
leaderElectionDriver = createLeaderElectionDriver();
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java
index d101d19b805..4555925b10f 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java
@@ -42,7 +42,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -84,9 +83,7 @@ class KubernetesLeaderElectionAndRetrievalITCase {
final List<AutoCloseable> closeables = new ArrayList<>();
final KubernetesConfigMapSharedWatcher configMapSharedWatcher =
- flinkKubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
+ flinkKubeClient.createConfigMapSharedWatcher(configMapName);
closeables.add(configMapSharedWatcher);
final TestingLeaderElectionEventHandler electionEventHandler =
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesTestFixture.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesTestFixture.java
index 9552ede3185..afbfd4ec0f4 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesTestFixture.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesTestFixture.java
@@ -26,7 +26,6 @@ import org.apache.flink.kubernetes.kubeclient.TestingFlinkKubeClient;
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.utils.KubernetesUtils;
import org.apache.flink.util.concurrent.FutureUtils;
import java.util.ArrayList;
@@ -38,7 +37,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.kubernetes.kubeclient.resources.KubernetesLeaderElector.LEADER_ANNOTATION_KEY;
-import static org.apache.flink.kubernetes.utils.Constants.LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY;
import static org.assertj.core.api.Assertions.assertThat;
/** Test fixture for Kubernetes tests that sets up a mock {@link FlinkKubeClient}. */
@@ -58,8 +56,6 @@ class KubernetesTestFixture {
private final List<TestingFlinkKubeClient.MockKubernetesWatch> configMapWatches =
new ArrayList<>();
- private final CompletableFuture<Map<String, String>> deleteConfigMapByLabelsFuture =
- new CompletableFuture<>();
private final CompletableFuture<Void> closeKubeClientFuture = new CompletableFuture<>();
private final CompletableFuture<KubernetesLeaderElector.LeaderCallbackHandler>
@@ -76,10 +72,7 @@ class KubernetesTestFixture {
configuration.setString(KubernetesConfigOptions.CLUSTER_ID, clusterId);
flinkKubeClient = createFlinkKubeClient();
- configMapSharedWatcher =
- flinkKubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
+ configMapSharedWatcher = flinkKubeClient.createConfigMapSharedWatcher(leaderConfigmapName);
}
void close() {
@@ -94,10 +87,6 @@ class KubernetesTestFixture {
return closeKubeClientFuture;
}
- CompletableFuture<Map<String, String>> getDeleteConfigMapByLabelsFuture() {
- return deleteConfigMapByLabelsFuture;
- }
-
KubernetesConfigMapSharedWatcher getConfigMapSharedWatcher() {
return configMapSharedWatcher;
}
@@ -178,19 +167,6 @@ class KubernetesTestFixture {
configMapStore.remove(name);
return FutureUtils.completedVoidFuture();
})
- .setDeleteConfigMapByLabelFunction(
- labels -> {
- if (deleteConfigMapByLabelsFuture.isDone()) {
- return FutureUtils.completedExceptionally(
- new KubernetesException(
- "ConfigMap with labels "
- + labels
- + " has already be deleted."));
- } else {
- deleteConfigMapByLabelsFuture.complete(labels);
- return FutureUtils.completedVoidFuture();
- }
- })
.setCloseConsumer(closeKubeClientFuture::complete)
.setCreateLeaderElectorFunction(
(leaderConfig, callbackHandler) -> {
@@ -199,12 +175,11 @@ class KubernetesTestFixture {
leaderConfig, callbackHandler);
})
.setCreateConfigMapSharedWatcherFunction(
- (labels) -> {
+ name -> {
final TestingFlinkKubeClient.TestingKubernetesConfigMapSharedWatcher
watcher =
new TestingFlinkKubeClient
- .TestingKubernetesConfigMapSharedWatcher(
- labels);
+ .TestingKubernetesConfigMapSharedWatcher(name);
watcher.setWatchFunction(
(ignore, handler) -> {
final CompletableFuture<
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
index 322a8af0415..6530a0742f8 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java
@@ -447,21 +447,6 @@ public class Fabric8FlinkKubeClientTest extends KubernetesClientTestBase {
.containsEntry(TESTING_CONFIG_MAP_KEY, TESTING_CONFIG_MAP_VALUE);
}
- @Test
- void testDeleteConfigMapByLabels() throws Exception {
- this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
- assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME)).isPresent();
- this.flinkKubeClient.deleteConfigMapsByLabels(TESTING_LABELS).get();
- assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME)).isNotPresent();
- }
-
- @Test
- void testDeleteNotExistingConfigMapByLabels() throws Exception {
- assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME)).isNotPresent();
- this.flinkKubeClient.deleteConfigMapsByLabels(TESTING_LABELS).get();
- assertThat(this.flinkKubeClient.getConfigMap(TESTING_CONFIG_MAP_NAME)).isNotPresent();
- }
-
@Test
void testDeleteConfigMapByName() throws Exception {
this.flinkKubeClient.createConfigMap(buildTestingConfigMap()).get();
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java
index be67ffa7eea..c3cde1e0927 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java
@@ -61,8 +61,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
Function<KubernetesConfigMap, Optional<KubernetesConfigMap>>,
CompletableFuture<Boolean>>
checkAndUpdateConfigMapFunction;
- private final Function<Map<String, String>, CompletableFuture<Void>>
- deleteConfigMapByLabelFunction;
private final Function<String, CompletableFuture<Void>> deleteConfigMapFunction;
private final Consumer<Void> closeConsumer;
private final BiFunction<
@@ -71,7 +69,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
KubernetesLeaderElector>
createLeaderElectorFunction;
- private final Function<Map<String, String>, KubernetesConfigMapSharedWatcher>
+ private final Function<String, KubernetesConfigMapSharedWatcher>
createConfigMapSharedWatcherFunction;
private TestingFlinkKubeClient(
@@ -88,7 +86,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
Function<KubernetesConfigMap, Optional<KubernetesConfigMap>>,
CompletableFuture<Boolean>>
checkAndUpdateConfigMapFunction,
- Function<Map<String, String>, CompletableFuture<Void>> deleteConfigMapByLabelFunction,
Function<String, CompletableFuture<Void>> deleteConfigMapFunction,
Consumer<Void> closeConsumer,
BiFunction<
@@ -96,7 +93,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
KubernetesLeaderElector.LeaderCallbackHandler,
KubernetesLeaderElector>
createLeaderElectorFunction,
- Function<Map<String, String>, KubernetesConfigMapSharedWatcher>
+ Function<String, KubernetesConfigMapSharedWatcher>
createConfigMapSharedWatcherFunction) {
this.createTaskManagerPodFunction = createTaskManagerPodFunction;
@@ -108,7 +105,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
this.createConfigMapFunction = createConfigMapFunction;
this.getConfigMapFunction = getConfigMapFunction;
this.checkAndUpdateConfigMapFunction = checkAndUpdateConfigMapFunction;
- this.deleteConfigMapByLabelFunction = deleteConfigMapByLabelFunction;
this.deleteConfigMapFunction = deleteConfigMapFunction;
this.closeConsumer = closeConsumer;
@@ -182,20 +178,14 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
return checkAndUpdateConfigMapFunction.apply(configMapName, updateFunction);
}
- @Override
- public CompletableFuture<Void> deleteConfigMapsByLabels(Map<String, String> labels) {
- return deleteConfigMapByLabelFunction.apply(labels);
- }
-
@Override
public CompletableFuture<Void> deleteConfigMap(String configMapName) {
return deleteConfigMapFunction.apply(configMapName);
}
@Override
- public KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(
- Map<String, String> labels) {
- return createConfigMapSharedWatcherFunction.apply(labels);
+ public KubernetesConfigMapSharedWatcher createConfigMapSharedWatcher(String name) {
+ return createConfigMapSharedWatcherFunction.apply(name);
}
@Override
@@ -241,8 +231,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
CompletableFuture<Boolean>>
checkAndUpdateConfigMapFunction =
(ignore1, ignore2) -> CompletableFuture.completedFuture(true);
- private Function<Map<String, String>, CompletableFuture<Void>>
- deleteConfigMapByLabelFunction = (ignore) -> FutureUtils.completedVoidFuture();
private Function<String, CompletableFuture<Void>> deleteConfigMapFunction =
(ignore) -> FutureUtils.completedVoidFuture();
@@ -254,7 +242,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
KubernetesLeaderElector>
createLeaderElectorFunction = TestingKubernetesLeaderElector::new;
- private Function<Map<String, String>, KubernetesConfigMapSharedWatcher>
+ private Function<String, KubernetesConfigMapSharedWatcher>
createConfigMapSharedWatcherFunction = TestingKubernetesConfigMapSharedWatcher::new;
private Builder() {}
@@ -318,13 +306,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
return this;
}
- public Builder setDeleteConfigMapByLabelFunction(
- Function<Map<String, String>, CompletableFuture<Void>>
- deleteConfigMapByLabelFunction) {
- this.deleteConfigMapByLabelFunction = deleteConfigMapByLabelFunction;
- return this;
- }
-
public Builder setDeleteConfigMapFunction(
Function<String, CompletableFuture<Void>> deleteConfigMapFunction) {
this.deleteConfigMapFunction = deleteConfigMapFunction;
@@ -347,7 +328,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
}
public Builder setCreateConfigMapSharedWatcherFunction(
- Function<Map<String, String>, KubernetesConfigMapSharedWatcher>
+ Function<String, KubernetesConfigMapSharedWatcher>
createConfigMapSharedWatcherFunction) {
this.createConfigMapSharedWatcherFunction = createConfigMapSharedWatcherFunction;
return this;
@@ -363,7 +344,6 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
createConfigMapFunction,
getConfigMapFunction,
checkAndUpdateConfigMapFunction,
- deleteConfigMapByLabelFunction,
deleteConfigMapFunction,
closeConsumer,
createLeaderElectorFunction,
@@ -454,7 +434,7 @@ public class TestingFlinkKubeClient implements FlinkKubeClient {
private BiFunction<String, WatchCallbackHandler<KubernetesConfigMap>, Watch> watchFunction =
(ignore1, ignore2) -> new MockKubernetesWatch();
- public TestingKubernetesConfigMapSharedWatcher(Map<String, String> labels) {}
+ public TestingKubernetesConfigMapSharedWatcher(String name) {}
public void setWatchFunction(
BiFunction<String, WatchCallbackHandler<KubernetesConfigMap>, Watch>
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSharedInformerITCase.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSharedInformerITCase.java
index 3d3d5b3e02a..87d504a6997 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSharedInformerITCase.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/resources/KubernetesSharedInformerITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.kubeclient.resources;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.kubernetes.KubernetesExtension;
import org.apache.flink.kubernetes.kubeclient.FlinkKubeClient;
import org.apache.flink.kubernetes.kubeclient.KubernetesConfigMapSharedWatcher;
@@ -34,7 +35,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.RegisterExtension;
-import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -53,8 +53,7 @@ class KubernetesSharedInformerITCase {
@RegisterExtension
private static final KubernetesExtension kubernetesExtension = new KubernetesExtension();
- private final ImmutableMap<String, String> labels =
- ImmutableMap.of("app", "shared-informer-test-cluster");
+ private final String configMapName = "shared-informer-test-cluster";
private FlinkKubeClient client;
private ExecutorService watchCallbackExecutorService;
@@ -69,7 +68,7 @@ class KubernetesSharedInformerITCase {
@AfterEach
void tearDown() throws Exception {
ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, watchCallbackExecutorService);
- client.deleteConfigMapsByLabels(labels).get();
+ client.deleteConfigMap(configMapName).get();
}
@Test
@@ -77,48 +76,38 @@ class KubernetesSharedInformerITCase {
public void testWatch() throws Exception {
try (KubernetesConfigMapSharedWatcher watcher =
- client.createConfigMapSharedWatcher(labels)) {
- for (int i = 0; i < 10; i++) {
- List<TestingCallbackHandler> callbackHandlers = new ArrayList<>();
- List<Watch> watchers = new ArrayList<>();
-
- watchInRange(watcher, callbackHandlers, watchers, 0, 20);
- createConfigMapsInRange(0, 5);
- watchInRange(watcher, callbackHandlers, watchers, 20, 40);
- createConfigMapsInRange(5, 10);
- updateConfigMapInRange(0, 10, ImmutableMap.of("foo", "bar"));
- for (TestingCallbackHandler handler : callbackHandlers) {
- handler.addFuture.get();
- handler.addOrUpdateFuture.get();
- assertThat(handler.assertFuture).isNotCompletedExceptionally();
+ client.createConfigMapSharedWatcher(configMapName)) {
+ TestingCallbackHandler handler = new TestingCallbackHandler(configMapName);
+ try (Watch watch =
+ watcher.watch(configMapName, handler, watchCallbackExecutorService)) {
+ createConfigMap(configMapName);
+ FlinkAssertions.assertThatFuture(handler.addFuture)
+ .as("The creation of the ConfigMap should have been processed, eventually.")
+ .eventuallySucceeds();
+
+ updateConfigMap(configMapName, ImmutableMap.of("foo", "bar"));
+ FlinkAssertions.assertThatFuture(handler.addOrUpdateFuture)
+ .as("The update of the ConfigMap should have been processed, eventually.")
+ .eventuallySucceeds();
+ assertThat(handler.assertFuture).isNotCompletedExceptionally();
+
+ client.deleteConfigMap(configMapName).get();
+ FlinkAssertions.assertThatFuture(handler.deleteFuture)
+ .as("The deletion of the ConfigMap should have been processed, eventually.")
+ .eventuallySucceeds();
+ if (handler.assertFuture.isCompletedExceptionally()) {
+ handler.assertFuture.get();
}
- watchers.forEach(Watch::close);
- callbackHandlers.clear();
- watchers.clear();
-
- watchInRange(watcher, callbackHandlers, watchers, 40, 60);
- for (TestingCallbackHandler handler : callbackHandlers) {
- handler.addFuture.get();
- handler.addOrUpdateFuture.get();
- }
- client.deleteConfigMapsByLabels(labels).get();
- for (TestingCallbackHandler handler : callbackHandlers) {
- handler.deleteFuture.get();
- if (handler.assertFuture.isCompletedExceptionally()) {
- handler.assertFuture.get();
- }
- }
- watchers.forEach(Watch::close);
}
}
}
@Test
void testWatchWithBlockHandler() throws Exception {
+ final String configMapName = getConfigMapName(System.currentTimeMillis());
try (KubernetesConfigMapSharedWatcher watcher =
- client.createConfigMapSharedWatcher(labels)) {
+ client.createConfigMapSharedWatcher(configMapName)) {
- final String configMapName = getConfigMapName(System.currentTimeMillis());
final long block = 500;
final long maxUpdateVal = 30;
final TestingBlockCallbackHandler handler =
@@ -140,32 +129,17 @@ class KubernetesSharedInformerITCase {
}
}
- private void createConfigMapsInRange(int start, int end) throws Exception {
- for (int i = start; i < end; i++) {
- createConfigMap(getConfigMapName(i));
- }
- }
-
private void createConfigMap(String name) throws Exception {
client.createConfigMap(
new KubernetesConfigMap(
new ConfigMapBuilder()
.withNewMetadata()
.withName(name)
- .withLabels(labels)
.endMetadata()
.build()))
.get();
}
- private void updateConfigMapInRange(int start, int end, Map<String, String> data)
- throws Exception {
- for (int i = start; i < end; i++) {
- final String configMapName = getConfigMapName(i);
- updateConfigMap(configMapName, data);
- }
- }
-
private void updateConfigMap(String configMapName, Map<String, String> data) throws Exception {
client.checkAndUpdateConfigMap(
configMapName,
@@ -176,22 +150,6 @@ class KubernetesSharedInformerITCase {
.get();
}
- private void watchInRange(
- KubernetesConfigMapSharedWatcher watcher,
- List<TestingCallbackHandler> callbackHandlers,
- List<Watch> watchers,
- int start,
- int end) {
-
- for (int i = start; i < end; i++) {
- final String name = getConfigMapName(i % 10);
- final TestingCallbackHandler handler = new TestingCallbackHandler(name);
- callbackHandlers.add(handler);
- final Watch watch = watcher.watch(name, handler, watchCallbackExecutorService);
- watchers.add(watch);
- }
- }
-
private String getConfigMapName(long id) {
return "shared-informer-test-cluster-" + id;
}
(flink) 01/02: [FLINK-31385] Introduce extended Assertj matchers for completable futures, that don't rely on timeouts.
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2a83c910eee711f8b5f9dd4697de60221f21fb9d
Author: David Moravek <dm...@apache.org>
AuthorDate: Mon Feb 27 18:46:58 2023 +0100
[FLINK-31385] Introduce extended Assertj matchers for completable futures, that don't rely on timeouts.
Signed-off-by: David Moravek <dm...@apache.org>
---
.../flink/core/testutils/FlinkAssertions.java | 27 ++++
.../testutils/FlinkCompletableFutureAssert.java | 136 +++++++++++++++++++++
.../flink/core/testutils/FlinkAssertionsTest.java | 121 ++++++++++++++++++
3 files changed, 284 insertions(+)
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
index c2f51b0295a..ea11627cdf1 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
@@ -25,6 +25,8 @@ import org.assertj.core.api.InstanceOfAssertFactory;
import org.assertj.core.api.ListAssert;
import org.assertj.core.api.ThrowingConsumer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -139,4 +141,29 @@ public final class FlinkAssertions {
}
return Stream.concat(Stream.of(throwable), chainOfCauses(throwable.getCause()));
}
+
+ /**
+ * Create assertion for {@link java.util.concurrent.CompletableFuture}.
+ *
+ * @param actual the actual value.
+ * @param <T> the type of the value contained in the {@link
+ * java.util.concurrent.CompletableFuture}.
+ * @return the created assertion object.
+ */
+ public static <T> FlinkCompletableFutureAssert<T> assertThatFuture(
+ CompletableFuture<T> actual) {
+ return new FlinkCompletableFutureAssert<>(actual);
+ }
+
+ /**
+ * Create assertion for {@link java.util.concurrent.CompletionStage}.
+ *
+ * @param actual the actual value.
+ * @param <T> the type of the value contained in the {@link
+ * java.util.concurrent.CompletionStage}.
+ * @return the created assertion object.
+ */
+ public static <T> FlinkCompletableFutureAssert<T> assertThatFuture(CompletionStage<T> actual) {
+ return new FlinkCompletableFutureAssert<>(actual);
+ }
}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java
new file mode 100644
index 00000000000..c0833daa9ea
--- /dev/null
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java
@@ -0,0 +1,136 @@
+/*
+ * 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.core.testutils;
+
+import org.assertj.core.api.AbstractCompletableFutureAssert;
+import org.assertj.core.api.AssertionInfo;
+import org.assertj.core.api.ObjectAssert;
+import org.assertj.core.api.ThrowableAssertAlternative;
+import org.assertj.core.error.BasicErrorMessageFactory;
+import org.assertj.core.internal.Failures;
+import org.assertj.core.internal.Objects;
+
+import java.time.Duration;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Enhanced version of {@link org.assertj.core.api.CompletableFutureAssert}, that allows asserting
+ * futures without relying on timeouts.
+ *
+ * @param <T> type of the value contained in the {@link CompletableFuture}.
+ */
+public class FlinkCompletableFutureAssert<T>
+ extends AbstractCompletableFutureAssert<FlinkCompletableFutureAssert<T>, T> {
+
+ private static final String SHOULD_HAVE_SUCCEEDED = "%nExpecting%n <%s>%nto have succeeded.%n";
+
+ private static final String SHOULD_HAVE_FAILED = "%nExpecting%n <%s>%nto have failed.%n";
+
+ /** A strongly typed alternative to {@link org.assertj.core.api.WithThrowable}. */
+ public static class WithThrowable {
+
+ private final Throwable throwable;
+
+ private WithThrowable(Throwable throwable) {
+ this.throwable = throwable;
+ }
+
+ /**
+ * Checks that the underlying throwable is of the given type and returns a {@link
+ * ThrowableAssertAlternative} to chain further assertions on the underlying throwable.
+ *
+ * @param type the expected {@link Throwable} type
+ * @param <T> the expected {@link Throwable} type
+ * @return a {@link ThrowableAssertAlternative} built with underlying throwable.
+ */
+ public <T extends Throwable> ThrowableAssertAlternative<T> withThrowableOfType(
+ Class<T> type) {
+ final ThrowableAssertAlternative<Throwable> throwableAssert =
+ new ThrowableAssertAlternative<>(throwable).isInstanceOf(type);
+ @SuppressWarnings("unchecked")
+ final ThrowableAssertAlternative<T> cast =
+ (ThrowableAssertAlternative<T>) throwableAssert;
+ return cast;
+ }
+ }
+
+ FlinkCompletableFutureAssert(CompletableFuture<T> actual) {
+ super(actual, FlinkCompletableFutureAssert.class);
+ }
+
+ FlinkCompletableFutureAssert(CompletionStage<T> actual) {
+ super(actual.toCompletableFuture(), FlinkCompletableFutureAssert.class);
+ }
+
+ /**
+ * An equivalent of {@link #succeedsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @return a new assertion object on the future's result
+ */
+ public ObjectAssert<T> eventuallySucceeds() {
+ final T object = assertEventuallySucceeds(info, actual);
+ return new ObjectAssert<>(object);
+ }
+
+ /**
+ * An equivalent of {@link #failsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @return a new assertion instance on the future's exception.
+ */
+ public WithThrowable eventuallyFails() {
+ return new WithThrowable(assertEventuallyFails(info, actual));
+ }
+
+ /**
+ * An equivalent of {@link #failsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @param exceptionClass type of the exception we expect the future to complete with
+ * @return a new assertion instance on the future's exception.
+ * @param <E> type of the exception we expect the future to complete with
+ */
+ public <E extends Throwable> ThrowableAssertAlternative<E> eventuallyFailsWith(
+ Class<E> exceptionClass) {
+ return eventuallyFails().withThrowableOfType(exceptionClass);
+ }
+
+ private T assertEventuallySucceeds(AssertionInfo info, Future<T> actual) {
+ Objects.instance().assertNotNull(info, actual);
+ try {
+ return actual.get();
+ } catch (InterruptedException | ExecutionException | CancellationException e) {
+ throw Failures.instance()
+ .failure(info, new BasicErrorMessageFactory(SHOULD_HAVE_SUCCEEDED, actual));
+ }
+ }
+
+ private Exception assertEventuallyFails(AssertionInfo info, Future<?> actual) {
+ Objects.instance().assertNotNull(info, actual);
+ try {
+ actual.get();
+ throw Failures.instance()
+ .failure(info, new BasicErrorMessageFactory(SHOULD_HAVE_FAILED, actual));
+ } catch (InterruptedException | ExecutionException | CancellationException e) {
+ return e;
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java
new file mode 100644
index 00000000000..97ae1a72511
--- /dev/null
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.core.testutils;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test suite for {@link FlinkAssertions}. */
+class FlinkAssertionsTest {
+
+ @Test
+ void testEventuallySucceedsWithCompletedFuture() {
+ final CompletableFuture<String> completedFuture =
+ CompletableFuture.completedFuture("Apache Flink");
+ assertThatFuture(completedFuture).eventuallySucceeds().isEqualTo("Apache Flink");
+ }
+
+ @Test
+ void testEventuallySucceedsWithFailedFuture() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException("Squirrel"));
+ assertThatThrownBy(() -> assertThatFuture(failedFuture).eventuallySucceeds())
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Squirrel");
+ }
+
+ @Test
+ void testEventuallySucceedsWithIncompleteFutureTimesOut() throws Exception {
+ assertWaitingForInterrupt(
+ () ->
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(new CompletableFuture<>())
+ .eventuallySucceeds())
+ .isInstanceOf(AssertionError.class),
+ Duration.ofMillis(10));
+ }
+
+ @Test
+ void testEventuallyFailsWithCompletedFuture() {
+ final CompletableFuture<String> completedFuture =
+ CompletableFuture.completedFuture("Apache Flink");
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(completedFuture)
+ .eventuallyFailsWith(IllegalStateException.class))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Apache Flink");
+ }
+
+ @Test
+ void testEventuallyFailsWithFailedFuture() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException());
+ assertThatFuture(failedFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void testEventuallyFailsWithFailedFutureWithDifferentException() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException());
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(failedFuture)
+ .eventuallyFailsWith(CancellationException.class))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("CancellationException");
+ }
+
+ @Test
+ void testEventuallyFailsWithIncompleteFutureTimesOut() throws Exception {
+ assertWaitingForInterrupt(
+ () ->
+ assertThatFuture(new CompletableFuture<>())
+ .eventuallyFails()
+ .withThrowableOfType(InterruptedException.class),
+ Duration.ofMillis(10));
+ }
+
+ private static void assertWaitingForInterrupt(Runnable runnable, Duration timeout)
+ throws Exception {
+ final CheckedThread thread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ runnable.run();
+ }
+ };
+ thread.start();
+ Thread.sleep(timeout.toMillis());
+ assertThat(thread.isAlive()).isTrue();
+ thread.interrupt();
+ thread.sync();
+ }
+}