[FLINK-36876] Add proxy for restClient

Author: chenyuzhi

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java

@@ -53,6 +53,7 @@ public class FlinkOperatorConfiguration {
     Duration progressCheckInterval;
     Duration restApiReadyDelay;
     Duration flinkClientTimeout;
+    int flinkClientIOThreads;
     String flinkServiceHostOverride;
     Set<String> watchedNamespaces;
     boolean dynamicNamespacesEnabled;
@@ -99,6 +100,10 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
         Duration flinkClientTimeout =
                 operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_TIMEOUT);
 
+        int flinkClientIOThreads =
+                operatorConfig.getInteger(
+                        KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_IO_THREADS);
+
         Duration flinkCancelJobTimeout =
                 operatorConfig.get(
                         KubernetesOperatorConfigOptions.OPERATOR_FLINK_CLIENT_CANCEL_TIMEOUT);
@@ -209,6 +214,7 @@ public static FlinkOperatorConfiguration fromConfiguration(Configuration operato
                 progressCheckInterval,
                 restApiReadyDelay,
                 flinkClientTimeout,
+                flinkClientIOThreads,
                 flinkServiceHostOverride,
                 watchedNamespaces,
                 dynamicNamespacesEnabled,

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java

@@ -115,6 +115,14 @@ public static String operatorConfigKey(String key) {
                     .withDescription(
                             "The timeout for the observer to wait the flink rest client to return.");
 
+    @Documentation.Section(SECTION_SYSTEM)
+    public static final ConfigOption<Integer> OPERATOR_FLINK_CLIENT_IO_THREADS =
+            operatorConfig("kubernetes.operator.flink.client.io.threads")
+                    .intType()
+                    .defaultValue(60)
+                    .withDescription(
+                            "The maximum number of io threads used by the flink rest client.");
+
     @Documentation.Section(SECTION_SYSTEM)
     public static final ConfigOption<Duration> OPERATOR_FLINK_CLIENT_CANCEL_TIMEOUT =
             operatorConfig("flink.client.cancel.timeout")

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java

@@ -24,6 +24,8 @@
 import org.apache.flink.autoscaler.config.AutoScalerOptions;
 import org.apache.flink.autoscaler.utils.JobStatusUtils;
 import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.client.program.rest.retry.ExponentialWaitStrategy;
+import org.apache.flink.client.program.rest.retry.WaitStrategy;
 import org.apache.flink.configuration.CheckpointingOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.RestOptions;
@@ -56,6 +58,7 @@
 import org.apache.flink.kubernetes.operator.utils.ExceptionUtils;
 import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
 import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.highavailability.ClientHighAvailabilityServicesFactory;
 import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
 import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
@@ -109,6 +112,7 @@
 import org.apache.flink.util.Preconditions;
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
 
 import io.fabric8.kubernetes.api.model.DeletionPropagation;
@@ -129,6 +133,7 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.lang.reflect.Constructor;
 import java.net.HttpURLConnection;
 import java.net.InetSocketAddress;
 import java.net.MalformedURLException;
@@ -175,17 +180,20 @@ public abstract class AbstractFlinkService implements FlinkService {
     protected final ExecutorService executorService;
     protected final FlinkOperatorConfiguration operatorConfig;
     protected final ArtifactManager artifactManager;
+    private final EventLoopGroup flinkClientSharedEventLoopGroup;
     private static final String EMPTY_JAR = createEmptyJar();
 
     public AbstractFlinkService(
             KubernetesClient kubernetesClient,
             ArtifactManager artifactManager,
             ExecutorService executorService,
-            FlinkOperatorConfiguration operatorConfig) {
+            FlinkOperatorConfiguration operatorConfig,
+            EventLoopGroup flinkClientEventLoopGroup) {
         this.kubernetesClient = kubernetesClient;
         this.artifactManager = artifactManager;
         this.executorService = executorService;
         this.operatorConfig = operatorConfig;
+        this.flinkClientSharedEventLoopGroup = flinkClientEventLoopGroup;
     }
 
     protected abstract PodList getJmPodList(String namespace, String clusterId);
@@ -842,11 +850,34 @@ public RestClusterClient<String> getClusterClient(Configuration conf) throws Exc
                         operatorConfig.getFlinkServiceHostOverride(),
                         ExternalServiceDecorator.getNamespacedExternalServiceName(
                                 clusterId, namespace));
+
         final String restServerAddress = String.format("http://%s:%s", host, port);
-        return new RestClusterClient<>(
-                operatorRestConf,
-                clusterId,
-                (c, e) -> new StandaloneClientHAServices(restServerAddress));
+        RestClient restClient =
+                new RestClientProxy(
+                        operatorRestConf,
+                        executorService,
+                        host,
+                        port,
+                        flinkClientSharedEventLoopGroup);
+
+        Constructor<?> clusterClientConstructor =
+                RestClusterClient.class.getDeclaredConstructor(
+                        Configuration.class,
+                        RestClient.class,
+                        Object.class,
+                        WaitStrategy.class,
+                        ClientHighAvailabilityServicesFactory.class);
+
+        clusterClientConstructor.setAccessible(true);
+
+        return (RestClusterClient<String>)
+                clusterClientConstructor.newInstance(
+                        operatorRestConf,
+                        restClient,
+                        clusterId,
+                        new ExponentialWaitStrategy(10L, 2000L),
+                        (ClientHighAvailabilityServicesFactory)
+                                (c, e) -> new StandaloneClientHAServices(restServerAddress));
     }
 
     @Override

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java

@@ -36,6 +36,8 @@
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
 
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -74,6 +76,7 @@ public static final class ExceptionCacheEntry {
 
     protected final Map<Tuple2<Class<?>, ResourceID>, KubernetesResourceMetricGroup>
             resourceMetricGroups = new ConcurrentHashMap<>();
+    private final EventLoopGroup flinkClientEventLoopGroup;
 
     public FlinkResourceContextFactory(
             FlinkConfigManager configManager,
@@ -87,6 +90,10 @@ public FlinkResourceContextFactory(
                 Executors.newFixedThreadPool(
                         configManager.getOperatorConfiguration().getReconcilerMaxParallelism(),
                         new ExecutorThreadFactory("Flink-RestClusterClient-IO"));
+        this.flinkClientEventLoopGroup =
+                new NioEventLoopGroup(
+                        configManager.getOperatorConfiguration().getFlinkClientIOThreads(),
+                        new ExecutorThreadFactory("flink-rest-client-netty-shared"));
     }
 
     public FlinkStateSnapshotContext getFlinkStateSnapshotContext(
@@ -154,13 +161,15 @@ protected FlinkService getFlinkService(FlinkResourceContext<?> ctx) {
                         artifactManager,
                         clientExecutorService,
                         ctx.getOperatorConfig(),
-                        eventRecorder);
+                        eventRecorder,
+                        flinkClientEventLoopGroup);
             case STANDALONE:
                 return new StandaloneFlinkService(
                         ctx.getKubernetesClient(),
                         artifactManager,
                         clientExecutorService,
-                        ctx.getOperatorConfig());
+                        ctx.getOperatorConfig(),
+                        flinkClientEventLoopGroup);
             default:
                 throw new UnsupportedOperationException(
                         String.format("Unsupported deployment mode: %s", deploymentMode));

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java

@@ -51,6 +51,8 @@
 import org.apache.flink.runtime.rest.messages.job.JobResourceRequirementsHeaders;
 import org.apache.flink.runtime.rest.messages.job.JobResourcesRequirementsUpdateHeaders;
 
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+
 import io.fabric8.kubernetes.api.model.DeletionPropagation;
 import io.fabric8.kubernetes.api.model.PodList;
 import io.fabric8.kubernetes.api.model.apps.Deployment;
@@ -89,7 +91,28 @@ public NativeFlinkService(
             ExecutorService executorService,
             FlinkOperatorConfiguration operatorConfig,
             EventRecorder eventRecorder) {
-        super(kubernetesClient, artifactManager, executorService, operatorConfig);
+        this(
+                kubernetesClient,
+                artifactManager,
+                executorService,
+                operatorConfig,
+                eventRecorder,
+                null);
+    }
+
+    public NativeFlinkService(
+            KubernetesClient kubernetesClient,
+            ArtifactManager artifactManager,
+            ExecutorService executorService,
+            FlinkOperatorConfiguration operatorConfig,
+            EventRecorder eventRecorder,
+            EventLoopGroup flinkClientEventLoopGroup) {
+        super(
+                kubernetesClient,
+                artifactManager,
+                executorService,
+                operatorConfig,
+                flinkClientEventLoopGroup);
         this.eventRecorder = eventRecorder;
     }
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/RestClientProxy.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.service;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.RestClient;
+import org.apache.flink.util.ConfigurationException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.AbstractBootstrap;
+import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap;
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+/** Proxy for the {@link RestClient}. */
+public class RestClientProxy extends RestClient {
+    private static final Logger LOG = LoggerFactory.getLogger(RestClientProxy.class);
+
+    private final boolean useSharedEventLoopGroup;
+    private Field groupField;
+    private Bootstrap bootstrap;
+    private CompletableFuture<Void> terminationFuture;
+
+    public RestClientProxy(
+            Configuration configuration,
+            ExecutorService executor,
+            String host,
+            int port,
+            EventLoopGroup sharedGroup)
+            throws ConfigurationException, NoSuchFieldException, IllegalAccessException {
+        super(configuration, executor, host, port);
+
+        if (sharedGroup != null) {
+            Preconditions.checkArgument(
+                    !sharedGroup.isShuttingDown() && !sharedGroup.isShutdown(),
+                    "provided eventLoopGroup is shut/shutting down");
+
+            // get private field
+            Field bootstrapField = RestClient.class.getDeclaredField("bootstrap");
+            Field terminationFutureField = RestClient.class.getDeclaredField("terminationFuture");
+            Field groupField = AbstractBootstrap.class.getDeclaredField("group");
+
+            bootstrapField.setAccessible(true);
+            terminationFutureField.setAccessible(true);
+            groupField.setAccessible(true);
+
+            // TODO check null
+            this.terminationFuture = (CompletableFuture<Void>) terminationFutureField.get(this);
+            this.bootstrap = (Bootstrap) bootstrapField.get(this);
+            this.groupField = groupField;
+
+            // close previous group
+            bootstrap.config().group().shutdown();
+            // setup share group
+            groupField.set(bootstrap, sharedGroup);
+
+            useSharedEventLoopGroup = true;
+        } else {
+            useSharedEventLoopGroup = false;
+        }
+    }
+
+    @Override
+    public CompletableFuture<Void> closeAsync() {
+        if (useSharedEventLoopGroup) {
+            this.shutdownInternal();
+        }
+
+        return super.closeAsync();
+    }
+
+    @Override
+    public void shutdown(Time timeout) {
+        if (useSharedEventLoopGroup) {
+            this.shutdownInternal();
+        }
+        super.shutdown(timeout);
+    }
+
+    private void shutdownInternal() {
+        try {
+            // replace bootstrap's group to null to avoid shutdown shared group
+            groupField.set(bootstrap, null);
+            terminationFuture.complete(null);
+        } catch (IllegalAccessException e) {
+            LOG.error("Failed to setup rest client event group .", e);
+        }
+    }
+}

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java

@@ -39,6 +39,8 @@
 import org.apache.flink.kubernetes.operator.utils.StandaloneKubernetesUtils;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
+import org.apache.flink.shaded.netty4.io.netty.channel.EventLoopGroup;
+
 import io.fabric8.kubernetes.api.model.DeletionPropagation;
 import io.fabric8.kubernetes.api.model.PodList;
 import io.fabric8.kubernetes.client.KubernetesClient;
@@ -61,7 +63,21 @@ public StandaloneFlinkService(
             ArtifactManager artifactManager,
             ExecutorService executorService,
             FlinkOperatorConfiguration operatorConfig) {
-        super(kubernetesClient, artifactManager, executorService, operatorConfig);
+        this(kubernetesClient, artifactManager, executorService, operatorConfig, null);
+    }
+
+    public StandaloneFlinkService(
+            KubernetesClient kubernetesClient,
+            ArtifactManager artifactManager,
+            ExecutorService executorService,
+            FlinkOperatorConfiguration operatorConfig,
+            EventLoopGroup flinkClientEventLoopGroup) {
+        super(
+                kubernetesClient,
+                artifactManager,
+                executorService,
+                operatorConfig,
+                flinkClientEventLoopGroup);
     }
 
     @Override

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java

@@ -71,8 +71,10 @@
 import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedSubtaskMetricsHeaders;
 import org.apache.flink.runtime.rest.util.RestClientException;
 import org.apache.flink.util.SerializedThrowable;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 import org.apache.flink.util.concurrent.Executors;
 
+import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
 import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
 
 import io.fabric8.kubernetes.api.model.DeletionPropagation;
@@ -171,7 +173,9 @@ public TestingFlinkService(KubernetesClient kubernetesClient) {
                 kubernetesClient,
                 null,
                 Executors.newDirectExecutorService(),
-                FlinkOperatorConfiguration.fromConfiguration(new Configuration()));
+                FlinkOperatorConfiguration.fromConfiguration(new Configuration()),
+                new NioEventLoopGroup(
+                        4, new ExecutorThreadFactory("flink-rest-client-netty-shared")));
     }
 
     public <T extends HasMetadata> Context<T> getContext() {