xintongsong commented on a change in pull request #13755:
URL: https://github.com/apache/flink/pull/13755#discussion_r512363656
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterClientFactory.java
##########
@@ -41,6 +42,8 @@
private static final String CLUSTER_ID_PREFIX = "flink-cluster-";
+ private static final KubeClientFactory kubeClientFactory =
DefaultKubeClientFactory.getInstance();
Review comment:
minor: this can be a local variable
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java
##########
@@ -75,7 +80,7 @@
/** Current max pod index. When creating a new pod, it should increase
one. */
private long currentMaxPodId = 0;
- private KubernetesWatch podsWatch;
+ private Optional<KubernetesWatch> podsWatch;
Review comment:
The `podsWatch` related changes should be in a separate commit. They are
hotfix for exist codes and is unrelated to the purpose of this PR.
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesResourceManagerDriver.java
##########
@@ -158,7 +165,9 @@ public void deregisterApplication(ApplicationStatus
finalStatus, @Nullable Strin
// In case of pod creation failures, we should wait for an
interval before trying to create new pods.
// Otherwise, ActiveResourceManager will always re-requesting
the worker, which keeps the main thread busy.
final CompletableFuture<Void> createPodFuture =
- podCreationCoolDown.thenCompose((ignore) ->
kubeClient.createTaskManagerPod(taskManagerPod));
+ podCreationCoolDown.thenCompose((ignore) ->
kubeClient
+ .map(client ->
client.createTaskManagerPod(taskManagerPod))
+
.orElse(CompletableFuture.completedFuture(null)));
Review comment:
Since we are expecting the client to be non-empty, I think we should
fail explicitly for the empty case.
Same for the other occurrences.
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/TestingKubeClientFactory.java
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.kubeclient;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.util.ExecutorThreadFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+
+/**
+ * Testing implementation of {@link KubeClientFactory}.
+ */
+public class TestingKubeClientFactory implements KubeClientFactory {
+
+ private TestingFlinkKubeClient.Builder flinkKubeClientBuilder;
Review comment:
Should be `final`.
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java
##########
@@ -64,6 +65,8 @@
"stop - stop the kubernetes cluster\n" +
"quit - quit attach mode";
+ private final KubeClientFactory kubeClientFactory =
DefaultKubeClientFactory.getInstance();
Review comment:
minor: this can be a local variable
----------------------------------------------------------------
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:
us...@infra.apache.org
