zentol commented on a change in pull request #13190:
URL: https://github.com/apache/flink/pull/13190#discussion_r472915726
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java
##########
@@ -96,40 +97,51 @@ private int run(String[] args) throws FlinkException,
CliArgsException {
kubernetesClusterClientFactory.createClusterDescriptor(configuration);
try {
- final ClusterClient<String> clusterClient;
+ final ClusterClientProvider<String>
clusterClientProvider;
String clusterId =
kubernetesClusterClientFactory.getClusterId(configuration);
final boolean detached =
!configuration.get(DeploymentOptions.ATTACHED);
- final FlinkKubeClient kubeClient =
KubeClientFactory.fromConfiguration(configuration);
+
+ boolean connectToExistingCluster = false;
+ try (final FlinkKubeClient kubeClient =
KubeClientFactory.fromConfiguration(configuration)) {
+ connectToExistingCluster = clusterId != null &&
kubeClient.getRestService(clusterId).isPresent();
+ } catch (Exception e) {
+ LOG.info("Could not properly shutdown cluster
client.", e);
+ }
// Retrieve or create a session cluster.
- if (clusterId != null &&
kubeClient.getRestService(clusterId).isPresent()) {
- clusterClient =
kubernetesClusterDescriptor.retrieve(clusterId).getClusterClient();
+ if (connectToExistingCluster) {
+ clusterClientProvider =
kubernetesClusterDescriptor.retrieve(clusterId);
} else {
- clusterClient = kubernetesClusterDescriptor
+ clusterClientProvider =
kubernetesClusterDescriptor
.deploySessionCluster(
-
kubernetesClusterClientFactory.getClusterSpecification(configuration))
- .getClusterClient();
- clusterId = clusterClient.getClusterId();
+
kubernetesClusterClientFactory.getClusterSpecification(configuration));
}
- try {
- if (!detached) {
- Tuple2<Boolean, Boolean> continueRepl =
new Tuple2<>(true, false);
- try (BufferedReader in = new
BufferedReader(new InputStreamReader(System.in))) {
- while (continueRepl.f0) {
- continueRepl =
repStep(in);
- }
+ if (!detached) {
+ if (!connectToExistingCluster) {
Review comment:
```suggestion
if (clusterId == null) {
```
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/cli/KubernetesSessionCli.java
##########
@@ -96,40 +97,51 @@ private int run(String[] args) throws FlinkException,
CliArgsException {
kubernetesClusterClientFactory.createClusterDescriptor(configuration);
try {
- final ClusterClient<String> clusterClient;
+ final ClusterClientProvider<String>
clusterClientProvider;
String clusterId =
kubernetesClusterClientFactory.getClusterId(configuration);
final boolean detached =
!configuration.get(DeploymentOptions.ATTACHED);
- final FlinkKubeClient kubeClient =
KubeClientFactory.fromConfiguration(configuration);
+
+ boolean connectToExistingCluster = false;
+ try (final FlinkKubeClient kubeClient =
KubeClientFactory.fromConfiguration(configuration)) {
+ connectToExistingCluster = clusterId != null &&
kubeClient.getRestService(clusterId).isPresent();
+ } catch (Exception e) {
+ LOG.info("Could not properly shutdown cluster
client.", e);
+ }
// Retrieve or create a session cluster.
- if (clusterId != null &&
kubeClient.getRestService(clusterId).isPresent()) {
- clusterClient =
kubernetesClusterDescriptor.retrieve(clusterId).getClusterClient();
Review comment:
hmm....is it a hard requirement that a cluster-id must be specified when
starting a cluster, or can it also be auto-generated by Kubernetes?
If it is not required, then we could adjust the PR such that the cluster-id
is only queried if `clusterId` is null:
https://github.com/apache/flink/pull/13190/files#r472915726
----------------------------------------------------------------
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
