You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "zhougit86 (via GitHub)" <gi...@apache.org> on 2024/01/22 08:42:44 UTC
[PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
zhougit86 opened a new pull request, #24163:
URL: https://github.com/apache/flink/pull/24163
…ch fail
## What is the purpose of the change
prevent jm restart in case kubernetes fails to handle watch request
## Brief change log
use ExponentialBackoffRetryStrategy to retry watch
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: ( no )
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes )
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (docs)
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1914013482
@zhougit86 ,
If you need any help in understanding the comments, feel free to reach out to my email (tonysong820@gmail.com). We may exchange WeChat ID and have an offline phone call if you'd like to.
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1903516752
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "67741049ceefd714af4f21588c63d222104a9e88",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "67741049ceefd714af4f21588c63d222104a9e88",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 67741049ceefd714af4f21588c63d222104a9e88 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong closed pull request #24163: [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat…
URL: https://github.com/apache/flink/pull/24163
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "zhougit86 (via GitHub)" <gi...@apache.org>.
zhougit86 commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1956068736
> @zhougit86,
>
> Thanks for addressing my comments. The PR LGTM.
>
> The CI failures seem unrelated to this PR. I have just rebased the PR onto the latest master branch. Let's see if that resolves the CI failures.
>
> I also have another two minor comments, which I addressed myself in a fixup commit. Please check them out.
>
> If you have no objections on my changes, I'll merge this PR once it passes the CI.
Thanks for your advice, really appreciate!
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #24163:
URL: https://github.com/apache/flink/pull/24163#discussion_r1469101401
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java:
##########
@@ -454,6 +451,22 @@ void testWatchPodsAndDoCallback() throws Exception {
assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
}
+ @Test
+ void testWatchPodsAndDoCallbackFail() throws Exception {
+ mockPodEventWithLabelsFail(
+ NAMESPACE, TASKMANAGER_POD_NAME, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
+ TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
+ TestingWatchCallbackHandler.<KubernetesPod>builder().build();
+ Long watchStartTime = System.currentTimeMillis();
+ assertThatThrownBy(
+ () ->
+ this.flinkKubeClient.watchPodsAndDoCallback(
+ TESTING_LABELS, watchCallbackHandler));
Review Comment:
It is unclear to me whether `watchPodsAndDoCallback` suppose to succeed or fail? There's no verification against the error being thrown.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java:
##########
@@ -183,6 +183,29 @@ protected void mockPodEventWithLabels(
.once();
}
+ protected void mockPodEventWithLabelsFail(
+ String namespace, String podName, String resourceVersion, Map<String, String> labels) {
Review Comment:
`podName` is never used.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java:
##########
@@ -454,6 +451,22 @@ void testWatchPodsAndDoCallback() throws Exception {
assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
}
+ @Test
+ void testWatchPodsAndDoCallbackFail() throws Exception {
+ mockPodEventWithLabelsFail(
+ NAMESPACE, TASKMANAGER_POD_NAME, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
+ TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
+ TestingWatchCallbackHandler.<KubernetesPod>builder().build();
+ Long watchStartTime = System.currentTimeMillis();
+ assertThatThrownBy(
+ () ->
+ this.flinkKubeClient.watchPodsAndDoCallback(
+ TESTING_LABELS, watchCallbackHandler));
+ Long watchEndTime = System.currentTimeMillis();
+ Long watchDuration = watchEndTime - watchStartTime;
+ assertThat(watchDuration > 100);
Review Comment:
Relying on the duration to decide whether the retry has happened can be unstable. I'd suggest to mock the server in a way that replies failure messages for certain times following by a success message.
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java:
##########
@@ -247,11 +259,12 @@ public KubernetesWatch watchPodsAndDoCallback(
new KubernetesPodsWatcher(
podCallbackHandler))),
kubeClientExecutorService),
- maxRetryAttempts,
+ new ExponentialBackoffRetryStrategy(
+ maxRetryAttempts, initialRetryInterval, maxRetryInterval),
t ->
ExceptionUtils.findThrowable(t, KubernetesClientException.class)
.isPresent(),
- kubeClientExecutorService)
+ new ScheduledExecutorServiceAdapter(kubeClientExecutorService))
.get();
Review Comment:
This means `watchPodsAndDoCallback` is blocking and can wait for minutes to return. It should not longer be called on the RPC main thread.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java:
##########
@@ -454,6 +451,22 @@ void testWatchPodsAndDoCallback() throws Exception {
assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
}
+ @Test
+ void testWatchPodsAndDoCallbackFail() throws Exception {
Review Comment:
`Exception` is never thrown.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java:
##########
@@ -183,6 +183,29 @@ protected void mockPodEventWithLabels(
.once();
}
+ protected void mockPodEventWithLabelsFail(
+ String namespace, String podName, String resourceVersion, Map<String, String> labels) {
+ final Pod pod =
+ new PodBuilder()
+ .withNewMetadata()
+ .withNamespace(namespace)
+ .withName(podName)
+ .withLabels(labels)
+ .withResourceVersion("5668")
+ .endMetadata()
+ .build();
Review Comment:
It seems there's no need to create this `pod`.
##########
docs/layouts/shortcodes/generated/kubernetes_config_configuration.html:
##########
@@ -290,5 +290,17 @@
<td>Integer</td>
<td>Defines the number of Kubernetes transactional operation retries before the client gives up. For example, <code class="highlighter-rouge">FlinkKubeClient#checkAndUpdateConfigMap</code>.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.transactional-operation.initial-retry-delay</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Duration</td>
+ <td>Defines the initial duration of Kubernetes transactional operation retries after fail.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.transactional-operation.max-retry-delay</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Duration</td>
+ <td>Defines the max duration of Kubernetes transactional operation retries after fail.</td>
+ </tr>
Review Comment:
We need to re-generate this document. The CI failure is related to this.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesClientTestBase.java:
##########
@@ -183,6 +183,29 @@ protected void mockPodEventWithLabels(
.once();
}
+ protected void mockPodEventWithLabelsFail(
Review Comment:
This name does not reflect what this method really does. I'd suggest `mockWatchPodFailure`
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java:
##########
@@ -454,6 +451,22 @@ void testWatchPodsAndDoCallback() throws Exception {
assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
}
+ @Test
+ void testWatchPodsAndDoCallbackFail() throws Exception {
+ mockPodEventWithLabelsFail(
+ NAMESPACE, TASKMANAGER_POD_NAME, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
+ TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
+ TestingWatchCallbackHandler.<KubernetesPod>builder().build();
+ Long watchStartTime = System.currentTimeMillis();
Review Comment:
I'd suggest to use `System.nanoTime()`.
`System.currentTimeMillis()` can be updated at a larger granularity and is not suitable ms-level time counting.
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #24163:
URL: https://github.com/apache/flink/pull/24163#discussion_r1462693554
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java:
##########
@@ -102,6 +109,14 @@ public Fabric8FlinkKubeClient(
this.maxRetryAttempts =
flinkConfig.getInteger(
KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRIES);
+ this.initialRetryInterval =
+ flinkConfig.getOptional(
+ KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_INITIAL_RETRY_DEALY)
+ .orElse(KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_INITIAL_RETRY_DEALY.defaultValue());
+ this.maxRetryInterval =
+ flinkConfig.getOptional(
+ KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRY_DEALY)
+ .orElse(KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRY_DEALY.defaultValue());
Review Comment:
There's no need to use `getOptional().orElse()`. A simple `get()` should automatically apply the default value if not configured.
```suggestion
this.initialRetryInterval =
flinkConfig.get(
KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_INITIAL_RETRY_DEALY);
this.maxRetryInterval =
flinkConfig.get(
KubernetesConfigOptions.KUBERNETES_TRANSACTIONAL_OPERATION_MAX_RETRY_DEALY);
```
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java:
##########
@@ -440,6 +441,28 @@ public class KubernetesConfigOptions {
code("FlinkKubeClient#checkAndUpdateConfigMap"))
.build());
+ public static final ConfigOption<Duration> KUBERNETES_TRANSACTIONAL_OPERATION_INITIAL_RETRY_DEALY =
+ key("kubernetes.transactional-operation.initial-retry-delay")
+ .durationType()
+ .defaultValue(Duration.ofMillis(5000))
Review Comment:
I think the 5s default initial delay might be too large. I'd suggest maybe 50ms.
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java:
##########
@@ -440,6 +441,28 @@ public class KubernetesConfigOptions {
code("FlinkKubeClient#checkAndUpdateConfigMap"))
.build());
+ public static final ConfigOption<Duration> KUBERNETES_TRANSACTIONAL_OPERATION_INITIAL_RETRY_DEALY =
+ key("kubernetes.transactional-operation.initial-retry-delay")
+ .durationType()
+ .defaultValue(Duration.ofMillis(5000))
Review Comment:
That also means it would take 12 retries to reach the default 1m max delay. Maybe we can also increase the default value for max-retires to maybe 15 or 20 times?
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "zhougit86 (via GitHub)" <gi...@apache.org>.
zhougit86 commented on code in PR #24163:
URL: https://github.com/apache/flink/pull/24163#discussion_r1495227606
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -114,7 +116,8 @@ public KubernetesResourceManagerDriver(
@Override
protected void initializeInternal() throws Exception {
- podsWatchOpt = watchTaskManagerPods();
+ podsWatchOptFuture = watchTaskManagerPods();
+ podsWatchOptFuture.get();
Review Comment:
This is because the watchTaskManagerPods before is a sync method, I just want to keep it the same behavior as before.
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "zhougit86 (via GitHub)" <gi...@apache.org>.
zhougit86 commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1903769843
@xintongsong Would you please take a look and let me know your thinking
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "zhougit86 (via GitHub)" <gi...@apache.org>.
zhougit86 commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1954032190
@flinkbot run azure
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "zhougit86 (via GitHub)" <gi...@apache.org>.
zhougit86 commented on PR #24163:
URL: https://github.com/apache/flink/pull/24163#issuecomment-1913601244
@xintongsong Hi , I got a chance to modify my commit, would you please bother to review for another time?
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #24163:
URL: https://github.com/apache/flink/pull/24163#discussion_r1496940478
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -90,7 +92,7 @@ public class KubernetesResourceManagerDriver
/** Current max pod index. When creating a new pod, it should increase one. */
private long currentMaxPodId = 0;
- private Optional<KubernetesWatch> podsWatchOpt;
+ private CompletableFuture<KubernetesWatch> podsWatchOptFuture;
Review Comment:
```suggestion
private CompletableFuture<KubernetesWatch> podsWatchOptFuture =
FutureUtils.completedExceptionally(
new ResourceManagerException(
"KubernetesResourceManagerDriver is not initialized."));
```
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -450,17 +462,30 @@ public void handleError(Throwable throwable) {
if (throwable instanceof KubernetesTooOldResourceVersionException) {
getMainThreadExecutor()
.execute(
- () -> {
- if (running) {
- podsWatchOpt.ifPresent(KubernetesWatch::close);
- log.info("Creating a new watch on TaskManager pods.");
- try {
- podsWatchOpt = watchTaskManagerPods();
- } catch (Exception e) {
- getResourceEventHandler().onError(e);
- }
- }
- });
+ () ->
+ podsWatchOptFuture.whenCompleteAsync(
+ (KubernetesWatch watch, Throwable throwable1) -> {
+ if (running) {
+ try {
+ if (watch != null) {
+ watch.close();
+ }
+ } catch (Exception e) {
+ log.warn(
+ "Error when get old watch to close, which is not supposed to happen",
+ e);
+ }
+ log.info(
+ "Creating a new watch on TaskManager pods.");
+ try {
+ podsWatchOptFuture =
+ watchTaskManagerPods();
+ } catch (Exception e) {
+ getResourceEventHandler().onError(e);
+ }
+ }
+ },
+ getMainThreadExecutor()));
Review Comment:
The outer `getMainThreadExecutor().execute()` is unnecessary.
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -90,7 +92,7 @@ public class KubernetesResourceManagerDriver
/** Current max pod index. When creating a new pod, it should increase one. */
private long currentMaxPodId = 0;
- private Optional<KubernetesWatch> podsWatchOpt;
+ private CompletableFuture<KubernetesWatch> podsWatchOptFuture;
Review Comment:
This helps prevent NPE if `podsWatchOptFuture` is accidentally being used before being initialized.
--
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
Re: [PR] [FLINK-33728] Do not rewatch when KubernetesResourceManagerDriver wat… [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #24163:
URL: https://github.com/apache/flink/pull/24163#discussion_r1494040971
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -412,11 +415,9 @@ private void stopPod(String podName) {
});
}
- private Optional<KubernetesWatch> watchTaskManagerPods() throws Exception {
- return Optional.of(
- flinkKubeClient.watchPodsAndDoCallback(
- KubernetesUtils.getTaskManagerSelectors(clusterId),
- new PodCallbackHandlerImpl()));
+ private CompletableFuture<KubernetesWatch> watchTaskManagerPods() throws Exception {
+ return flinkKubeClient.watchPodsAndDoCallback(
+ KubernetesUtils.getTaskManagerSelectors(clusterId), new PodCallbackHandlerImpl());
Review Comment:
It would be helpful to print a log when the future is completed.
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -114,7 +116,8 @@ public KubernetesResourceManagerDriver(
@Override
protected void initializeInternal() throws Exception {
- podsWatchOpt = watchTaskManagerPods();
+ podsWatchOptFuture = watchTaskManagerPods();
+ podsWatchOptFuture.get();
Review Comment:
Why do we need to wait for the `podsWatchOptFuture.get()` here? This blocks the RPC main thread.
##########
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java:
##########
@@ -452,10 +453,22 @@ public void handleError(Throwable throwable) {
.execute(
() -> {
if (running) {
- podsWatchOpt.ifPresent(KubernetesWatch::close);
+ try {
+ if (podsWatchOptFuture.get() != null) {
Review Comment:
This blocks the main thread. Is it necessary to wait for the previous watch to be closed before creating the new one? If yes, we should create the new watch in `podsWatchOptFuture.whenCompleteAsync()` to avoid blocking the main thread.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java:
##########
@@ -454,6 +451,21 @@ void testWatchPodsAndDoCallback() throws Exception {
assertThat(podModifiedAction.get()).isEqualTo(Action.MODIFIED);
}
+ @Test
+ void testWatchPodsAndDoCallbackFail() throws Exception {
+ mockWatchPodSuccessAfterFailTwoTimes(
+ NAMESPACE, KUBERNETES_ZERO_RESOURCE_VERSION, TESTING_LABELS);
+ TestingWatchCallbackHandler<KubernetesPod> watchCallbackHandler =
+ TestingWatchCallbackHandler.<KubernetesPod>builder().build();
+ // disable the retry of kubeClient
+ kubeClient.getConfiguration().setRequestRetryBackoffLimit(0);
+ assertThat(
+ flinkKubeClient
+ .watchPodsAndDoCallback(TESTING_LABELS, watchCallbackHandler)
+ .get())
Review Comment:
We should add a timeout, or the test case may hang for a long time.
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingFlinkKubeClient.java:
##########
@@ -149,9 +149,10 @@ public List<KubernetesPod> getPodsWithLabels(Map<String, String> labels) {
}
@Override
- public KubernetesWatch watchPodsAndDoCallback(
+ public CompletableFuture<KubernetesWatch> watchPodsAndDoCallback(
Map<String, String> labels, WatchCallbackHandler<KubernetesPod> podCallbackHandler) {
- return watchPodsAndDoCallbackFunction.apply(labels, podCallbackHandler);
+ return CompletableFuture.supplyAsync(
+ () -> watchPodsAndDoCallbackFunction.apply(labels, podCallbackHandler));
Review Comment:
I think we should change the return value type of `watchPodsAndDoCallbackFunction` to align with the method signature.
--
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