[GitHub] [flink] wangyang0918 commented on a diff in pull request #20673: [BP-1.15][FLINK-28265][k8s] Make KubernetesStateHandleStore#addEntry idempotent

[GitBox <gi...@apache.org>]

wangyang0918 commented on code in PR #20673:
URL: https://github.com/apache/flink/pull/20673#discussion_r955797916


##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesStateHandleStoreTest.java:
##########
@@ -1119,4 +1166,34 @@ private TestingLongStateHandleHelper.LongStateHandle addDeletingEntry(
         configMap.getData().put(key, deleting);
         return state;
     }
+
+    private static CompletableFuture<Boolean> retryWithFirstFailedK8sOperation(
+            Function<KubernetesConfigMap, Optional<KubernetesConfigMap>> function,
+            KubernetesConfigMap leaderConfigMap) {
+        final AtomicInteger callbackInvocationCount = new AtomicInteger(0);
+        final CompletableFuture<Boolean> result =
+                FutureUtils.retry(
+                        () ->
+                                CompletableFuture.supplyAsync(
+                                        () -> {
+                                            callbackInvocationCount.incrementAndGet();
+                                            function.apply(leaderConfigMap);
+                                            if (callbackInvocationCount.get() == 1) {
+                                                throw new KubernetesClientException(
+                                                        "Expected exception to simulate unstable "
+                                                                + "kubernetes client operation");
+                                            }
+                                            return true;
+                                        },
+                                        Executors.newDirectExecutorService()),
+                        KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES
+                                .defaultValue(),
+                        t ->
+                                ExceptionUtils.findThrowable(t, KubernetesClientException.class)
+                                        .isPresent(),
+                        Executors.newDirectExecutorService());
+        assertThat(callbackInvocationCount.get(), is(2));
+        assertThat(result.isDone(), is(true));

Review Comment:
   Since we are not using `assertj` in branch 1.15, do you think it is really necessary to also assert the exception here?
   
   ```
   assertThat(result.isCompletedExceptionally(), is(false));
   ```



-- 
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.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org