[GitHub] [flink] zhengcanbin commented on a change in pull request #11427: [FLINK-15790][k8s] Make FlinkKubeClient and its implementations asynchronous

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

zhengcanbin commented on a change in pull request #11427: [FLINK-15790][k8s] Make FlinkKubeClient and its implementations asynchronous
URL: https://github.com/apache/flink/pull/11427#discussion_r407877683
 
 

 ##########
 File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
 ##########
 @@ -61,158 +61,180 @@
 	private final String clusterId;
 	private final String nameSpace;
 
-	public Fabric8FlinkKubeClient(Configuration flinkConfig, KubernetesClient client) {
+	private final ExecutorWrapper executorWrapper;
+
+	public Fabric8FlinkKubeClient(Configuration flinkConfig, KubernetesClient client, ExecutorWrapper executorWrapper) {
 		this.flinkConfig = checkNotNull(flinkConfig);
 		this.internalClient = checkNotNull(client);
 		this.clusterId = checkNotNull(flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID));
 
 		this.nameSpace = flinkConfig.getString(KubernetesConfigOptions.NAMESPACE);
+
+		this.executorWrapper = executorWrapper;
 	}
 
 	@Override
-	public void createJobManagerComponent(KubernetesJobManagerSpecification kubernetesJMSpec) {
+	public CompletableFuture<Void> createJobManagerComponent(KubernetesJobManagerSpecification kubernetesJMSpec) {
 		final Deployment deployment = kubernetesJMSpec.getDeployment();
 		final List<HasMetadata> accompanyingResources = kubernetesJMSpec.getAccompanyingResources();
 
 		// create Deployment
 		LOG.debug("Start to create deployment with spec {}", deployment.getSpec().toString());
-		final Deployment createdDeployment = this.internalClient
-			.apps()
-			.deployments()
-			.inNamespace(this.nameSpace)
-			.create(deployment);
-
-		// Note that we should use the uid of the created Deployment for the OwnerReference.
-		setOwnerReference(createdDeployment, accompanyingResources);
 
-		this.internalClient
-			.resourceList(accompanyingResources)
-			.inNamespace(this.nameSpace)
-			.createOrReplace();
+		return CompletableFuture.runAsync(() -> {
+			final Deployment createdDeployment = this.internalClient
+				.apps()
+				.deployments()
+				.inNamespace(this.nameSpace)
+				.create(deployment);
+
+			// Note that we should use the uid of the created Deployment for the OwnerReference.
+			setOwnerReference(createdDeployment, accompanyingResources);
+
+			this.internalClient
+				.resourceList(accompanyingResources)
+				.inNamespace(this.nameSpace)
+				.createOrReplace();
+		}, executorWrapper.getExecutor());
 	}
 
 	@Override
 	public void createTaskManagerPod(KubernetesPod kubernetesPod) {
-		final Deployment masterDeployment = this.internalClient
-			.apps()
-			.deployments()
-			.inNamespace(this.nameSpace)
-			.withName(KubernetesUtils.getDeploymentName(clusterId))
-			.get();
-
-		if (masterDeployment == null) {
-			throw new RuntimeException(
-				"Failed to find Deployment named " + clusterId + " in namespace " + this.nameSpace);
-		}
+		CompletableFuture.runAsync(() -> {
 
 Review comment:
   > Alright, I think this should work in most cases. As a follow-up, we might want to make it configurable or add back-off (similar to `NETWORK_REQUEST_BACKOFF_INITIAL`/`NETWORK_REQUEST_BACKOFF_MAX`) if this causes problems of retrying not fast enough in the beginning or too fast.
   
   Thanks, @azagrebin ! Sorry to jump in, and this is a really good improvement, we have suffered such pains that the K8s API Server hangs after attacked by the infinite creation request from `KubernetesResourceManager`, the whole K8s cluster is unstable in such cases.

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services