[FLINK-15789][kubernetes] Do not wrap the InterruptedException and throw a...

[FLINK-15789][kubernetes] Do not wrap the InterruptedException and throw a different one in ActionWatcher.await

This closes #10970.

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) {

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);
 		}
 	}