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:31 UTC
(flink) 02/02: [FLINK-33598][k8s] Watch HA configmap via name instead of lables to reduce pressure on APIserver
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;
}