[flink] branch release-1.10 updated: [FLINK-15789][kubernetes] Do not wrap the InterruptedException and throw a different one in ActionWatcher.await

[tr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.10 by this push:
     new 9e45dd1  [FLINK-15789][kubernetes] Do not wrap the InterruptedException and throw a different one in ActionWatcher.await
9e45dd1 is described below

commit 9e45dd1be2227e83b6b98fdb64fdc274bd397e73
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Thu Jan 30 11:16:09 2020 +0800

    [FLINK-15789][kubernetes] Do not wrap the InterruptedException and throw a different one in ActionWatcher.await
    
    This closes #10970.
---
 .../flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java  | 12 +++++++-----
 .../flink/kubernetes/kubeclient/resources/ActionWatcher.java | 12 ++++--------
 2 files changed, 11 insertions(+), 13 deletions(-)

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 dd71e32..5d8d5e1 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
@@ -37,6 +37,7 @@ import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod;
 import org.apache.flink.kubernetes.kubeclient.resources.KubernetesService;
 import org.apache.flink.kubernetes.utils.Constants;
 import org.apache.flink.util.TimeUtils;
+import org.apache.flink.util.function.FunctionUtils;
 
 import io.fabric8.kubernetes.api.model.ConfigMap;
 import io.fabric8.kubernetes.api.model.Pod;
@@ -314,12 +315,13 @@ public class Fabric8FlinkKubeClient implements FlinkKubeClient {
 		final Duration timeout = TimeUtils.parseDuration(
 			flinkConfig.get(KubernetesConfigOptions.SERVICE_CREATE_TIMEOUT));
 
-		return CompletableFuture.supplyAsync(() -> {
-			final Service createdService = watcher.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
-			watchConnectionManager.close();
+		return CompletableFuture.supplyAsync(
+			FunctionUtils.uncheckedSupplier(() -> {
+				final Service createdService = watcher.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
+				watchConnectionManager.close();
 
-			return new KubernetesService(this.flinkConfig, createdService);
-		});
+				return new KubernetesService(this.flinkConfig, createdService);
+			}));
 	}
 
 	private KubernetesService getService(String serviceName) {
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java
index 89234e1..b5123ef 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/resources/ActionWatcher.java
@@ -53,14 +53,10 @@ public class ActionWatcher<T extends HasMetadata> implements Watcher<T> {
 	public void onClose(KubernetesClientException e) {
 	}
 
-	public T await(long amount, TimeUnit timeUnit) {
-		try {
-			if (this.latch.await(amount, timeUnit)) {
-				return this.reference.get();
-			} else {
-				throw new KubernetesClientTimeoutException(this.resource, amount, timeUnit);
-			}
-		} catch (InterruptedException var5) {
+	public T await(long amount, TimeUnit timeUnit) throws InterruptedException {
+		if (this.latch.await(amount, timeUnit)) {
+			return this.reference.get();
+		} else {
 			throw new KubernetesClientTimeoutException(this.resource, amount, timeUnit);
 		}
 	}