wangyang0918 commented on a change in pull request #13644:
URL: https://github.com/apache/flink/pull/13644#discussion_r507376641
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +230,68 @@ public KubernetesWatch watchPodsAndDoCallback(
.watch(new
KubernetesPodsWatcher(podCallbackHandler)));
}
+ @Override
+ public CompletableFuture<Void> createConfigMap(KubernetesConfigMap
configMap) {
+ return CompletableFuture.runAsync(
+ () -> {
+ if
(!getConfigMap(configMap.getName()).isPresent()) {
+
this.internalClient.configMaps().create(configMap.getInternalResource());
+ }
+ },
+ kubeClientExecutorService);
+ }
+
+ @Override
+ public Optional<KubernetesConfigMap> getConfigMap(String name) {
+ final ConfigMap configMap =
this.internalClient.configMaps().inNamespace(namespace).withName(name).get();
+ return configMap == null ? Optional.empty() : Optional.of(new
KubernetesConfigMap(configMap));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> checkAndUpdateConfigMap(
+ String configMapName,
+ Predicate<KubernetesConfigMap> checker,
+ FunctionWithException<KubernetesConfigMap,
KubernetesConfigMap, ?> function) {
+ return FutureUtils.retry(
+ () -> CompletableFuture.supplyAsync(
+ () -> getConfigMap(configMapName)
+
.map(FunctionUtils.uncheckedFunction(configMap -> {
+ final boolean shouldUpdate =
checker.test(configMap);
+ if (!shouldUpdate) {
+ LOG.warn("Trying to
update ConfigMap {} to {} without checking pass, ignoring.",
+
configMap.getName(), configMap.getData());
+ } else {
+
this.internalClient.configMaps()
+
.inNamespace(namespace)
+
.createOrReplace(function.apply(configMap).getInternalResource());
+ }
+ return shouldUpdate;
+ }))
+ .orElseThrow(
+ () -> new
FlinkRuntimeException("ConfigMap " + configMapName + " not exists.")),
+ kubeClientExecutorService),
+ maxRetryAttempts,
+ kubeClientExecutorService);
+ }
Review comment:
Yes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
