wangyang0918 commented on a change in pull request #13644:
URL: https://github.com/apache/flink/pull/13644#discussion_r511878028
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##########
@@ -219,6 +231,73 @@ public KubernetesWatch watchPodsAndDoCallback(
.watch(new
KubernetesPodsWatcher(podCallbackHandler)));
}
+ @Override
+ public CompletableFuture<Void> createConfigMap(KubernetesConfigMap
configMap) {
+ final String configMapName = configMap.getName();
+ return CompletableFuture.runAsync(
+ () ->
this.internalClient.configMaps().inNamespace(namespace).create(configMap.getInternalResource()),
+ kubeClientExecutorService)
+ .exceptionally(throwable -> {
+ if (throwable != null) {
+ throw new CompletionException(
+ new KubernetesException("Failed
to create ConfigMap " + configMapName, throwable));
+ }
+ return null;
+ });
+ }
+
+ @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,
+ FunctionWithException<KubernetesConfigMap,
Optional<KubernetesConfigMap>, ?> function) {
Review comment:
Copy the comment above.
We do not need the FunctionWithException here. Only the update operation
succeeded, we need to do the following external storage cleanup. For example,
we could use kubeClient.checkAndUpdateConfigMap(configMapName,
updateFunction).whenCompleteAsync() for this operation.
----------------------------------------------------------------
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]
