You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/04/12 09:22:09 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27141] Improve FlinkService#waitForClusterShutdown logic
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new a834352 [FLINK-27141] Improve FlinkService#waitForClusterShutdown logic
a834352 is described below
commit a83435212fabfcf82e7a7f26d455dc1f319090ca
Author: Nicholas Jiang <pr...@163.com>
AuthorDate: Tue Apr 12 17:22:04 2022 +0800
[FLINK-27141] Improve FlinkService#waitForClusterShutdown logic
---
docs/content/docs/operations/configuration.md | 2 ++
.../config/FlinkOperatorConfiguration.java | 15 ++++++++----
.../operator/config/OperatorConfigOptions.java | 12 ++++++++--
.../deployment/AbstractDeploymentReconciler.java | 6 ++++-
.../deployment/ApplicationReconciler.java | 6 ++++-
.../reconciler/deployment/SessionReconciler.java | 12 ++++++++--
.../kubernetes/operator/service/FlinkService.java | 27 +++++++++++++++------
.../kubernetes/operator/utils/FlinkUtils.java | 28 +++++++++++++++-------
.../kubernetes/operator/TestingFlinkService.java | 5 +++-
.../conf/flink-operator-config/flink-conf.yaml | 3 +++
10 files changed, 89 insertions(+), 27 deletions(-)
diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index a604839..96f22b9 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -36,3 +36,5 @@ under the License.
| operator.observer.progress-check.interval | 10s | Duration | The interval for observing status for in-progress operations such as deployment and savepoints. |
| operator.observer.savepoint.trigger.grace-period | 10s | Duration | The interval before a savepoint trigger attempt is marked as unsuccessful. |
| operator.observer.flink.client.timeout | 10s | Duration | The timeout for the observer to wait the flink rest client to return. |
+| operator.reconciler.flink.cancel.job.timeout | 1min | Duration | The timeout for the reconciler to wait for flink to cancel job. |
+| ooperator.reconciler.flink.cluster.shutdown.timeout | 60s | Duration | The timeout for the reconciler to wait for flink to shutdown cluster. |
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
index 4cea133..63ea937 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
@@ -39,7 +39,8 @@ public class FlinkOperatorConfiguration {
Duration flinkClientTimeout;
String flinkServiceHostOverride;
Set<String> watchedNamespaces;
- Duration cancelJobTimeout;
+ Duration flinkCancelJobTimeout;
+ Duration flinkShutdownClusterTimeout;
public static FlinkOperatorConfiguration fromConfiguration(Configuration operatorConfig) {
Set<String> watchedNamespaces = OperatorUtils.getWatchedNamespaces();
@@ -68,8 +69,13 @@ public class FlinkOperatorConfiguration {
Duration flinkClientTimeout =
operatorConfig.get(OperatorConfigOptions.OPERATOR_OBSERVER_FLINK_CLIENT_TIMEOUT);
- Duration cancelJobTimeout =
- operatorConfig.get(OperatorConfigOptions.OPERATOR_CANCEL_JOB_TIMEOUT);
+ Duration flinkCancelJobTimeout =
+ operatorConfig.get(
+ OperatorConfigOptions.OPERATOR_RECONCILER_FLINK_CANCEL_JOB_TIMEOUT);
+
+ Duration flinkShutdownClusterTimeout =
+ operatorConfig.get(
+ OperatorConfigOptions.OPERATOR_RECONCILER_FLINK_CLUSTER_SHUTDOWN_TIMEOUT);
String flinkServiceHostOverride = null;
if (EnvUtils.get("KUBERNETES_SERVICE_HOST") == null) {
@@ -86,6 +92,7 @@ public class FlinkOperatorConfiguration {
flinkClientTimeout,
flinkServiceHostOverride,
watchedNamespaces,
- cancelJobTimeout);
+ flinkCancelJobTimeout,
+ flinkShutdownClusterTimeout);
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
index 65d7cd3..48da537 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
@@ -70,9 +70,17 @@ public class OperatorConfigOptions {
.withDescription(
"The timeout for the observer to wait the flink rest client to return.");
- public static final ConfigOption<Duration> OPERATOR_CANCEL_JOB_TIMEOUT =
+ public static final ConfigOption<Duration> OPERATOR_RECONCILER_FLINK_CANCEL_JOB_TIMEOUT =
ConfigOptions.key("operator.reconciler.flink.cancel.job.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(1))
- .withDescription("The timeout for the operator to cancel job.");
+ .withDescription(
+ "The timeout for the reconciler to wait for flink to cancel job.");
+
+ public static final ConfigOption<Duration> OPERATOR_RECONCILER_FLINK_CLUSTER_SHUTDOWN_TIMEOUT =
+ ConfigOptions.key("operator.reconciler.flink.cluster.shutdown.timeout")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(60))
+ .withDescription(
+ "The timeout for the reconciler to wait for flink to shutdown cluster.");
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractDeploymentReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractDeploymentReconciler.java
index c0aced3..7f8056f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractDeploymentReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractDeploymentReconciler.java
@@ -58,7 +58,11 @@ public abstract class AbstractDeploymentReconciler implements Reconciler<FlinkDe
== flinkApp.getStatus().getJobManagerDeploymentStatus()) {
shutdown(flinkApp, effectiveConfig);
} else {
- FlinkUtils.deleteCluster(flinkApp, kubernetesClient, true);
+ FlinkUtils.deleteCluster(
+ flinkApp,
+ kubernetesClient,
+ true,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
}
return DeleteControl.defaultDelete();
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index e82859d..44277a3 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -231,7 +231,11 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
}
}
- FlinkUtils.deleteCluster(flinkApp, kubernetesClient, true);
+ FlinkUtils.deleteCluster(
+ flinkApp,
+ kubernetesClient,
+ true,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
}
private void triggerSavepoint(FlinkDeployment deployment, Configuration effectiveConfig)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index 9f3fa8f..a79688d 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -74,7 +74,11 @@ public class SessionReconciler extends AbstractDeploymentReconciler {
private void upgradeSessionCluster(FlinkDeployment flinkApp, Configuration effectiveConfig)
throws Exception {
LOG.info("Upgrading session cluster");
- flinkService.stopSessionCluster(flinkApp, effectiveConfig, false);
+ flinkService.stopSessionCluster(
+ flinkApp,
+ effectiveConfig,
+ false,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
flinkService.submitSessionCluster(flinkApp, effectiveConfig);
flinkApp.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
@@ -82,6 +86,10 @@ public class SessionReconciler extends AbstractDeploymentReconciler {
@Override
protected void shutdown(FlinkDeployment flinkApp, Configuration effectiveConfig) {
LOG.info("Stopping session cluster");
- flinkService.stopSessionCluster(flinkApp, effectiveConfig, true);
+ flinkService.stopSessionCluster(
+ flinkApp,
+ effectiveConfig,
+ true,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index 7cf22e0..6bc85b7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -284,7 +284,7 @@ public class FlinkService {
clusterClient
.cancel(jobID)
.get(
- operatorConfiguration.getCancelJobTimeout().toSeconds(),
+ operatorConfiguration.getFlinkCancelJobTimeout().toSeconds(),
TimeUnit.SECONDS);
break;
case SAVEPOINT:
@@ -313,13 +313,21 @@ public class FlinkService {
break;
case LAST_STATE:
final String namespace = conf.getString(KubernetesConfigOptions.NAMESPACE);
- FlinkUtils.deleteCluster(namespace, clusterId, kubernetesClient, false);
+ FlinkUtils.deleteCluster(
+ namespace,
+ clusterId,
+ kubernetesClient,
+ false,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
break;
default:
throw new RuntimeException("Unsupported upgrade mode " + upgradeMode);
}
}
- FlinkUtils.waitForClusterShutdown(kubernetesClient, conf);
+ FlinkUtils.waitForClusterShutdown(
+ kubernetesClient,
+ conf,
+ operatorConfiguration.getFlinkShutdownClusterTimeout().toSeconds());
return savepointOpt;
}
@@ -327,14 +335,19 @@ public class FlinkService {
try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
clusterClient
.cancel(jobID)
- .get(operatorConfiguration.getCancelJobTimeout().toSeconds(), TimeUnit.SECONDS);
+ .get(
+ operatorConfiguration.getFlinkCancelJobTimeout().toSeconds(),
+ TimeUnit.SECONDS);
}
}
public void stopSessionCluster(
- FlinkDeployment deployment, Configuration conf, boolean deleteHaData) {
- FlinkUtils.deleteCluster(deployment, kubernetesClient, deleteHaData);
- FlinkUtils.waitForClusterShutdown(kubernetesClient, conf);
+ FlinkDeployment deployment,
+ Configuration conf,
+ boolean deleteHaData,
+ long shutdownTimeout) {
+ FlinkUtils.deleteCluster(deployment, kubernetesClient, deleteHaData, shutdownTimeout);
+ FlinkUtils.waitForClusterShutdown(kubernetesClient, conf, shutdownTimeout);
}
public void triggerSavepoint(FlinkDeployment deployment, Configuration conf) throws Exception {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
index 1d89036..84ec9ef 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
@@ -130,12 +130,14 @@ public class FlinkUtils {
public static void deleteCluster(
FlinkDeployment flinkApp,
KubernetesClient kubernetesClient,
- boolean deleteHaConfigmaps) {
+ boolean deleteHaConfigmaps,
+ long shutdownTimeout) {
deleteCluster(
flinkApp.getMetadata().getNamespace(),
flinkApp.getMetadata().getName(),
kubernetesClient,
- deleteHaConfigmaps);
+ deleteHaConfigmaps,
+ shutdownTimeout);
}
/**
@@ -151,7 +153,8 @@ public class FlinkUtils {
String namespace,
String clusterId,
KubernetesClient kubernetesClient,
- boolean deleteHaConfigmaps) {
+ boolean deleteHaConfigmaps,
+ long shutdownTimeout) {
LOG.info("Deleting Flink cluster resources");
kubernetesClient
.apps()
@@ -163,7 +166,7 @@ public class FlinkUtils {
if (deleteHaConfigmaps) {
// We need to wait for cluster shutdown otherwise HA configmaps might be recreated
- waitForClusterShutdown(kubernetesClient, namespace, clusterId);
+ waitForClusterShutdown(kubernetesClient, namespace, clusterId, shutdownTimeout);
kubernetesClient
.configMaps()
.inNamespace(namespace)
@@ -176,12 +179,15 @@ public class FlinkUtils {
/** Wait until the FLink cluster has completely shut down. */
public static void waitForClusterShutdown(
- KubernetesClient kubernetesClient, String namespace, String clusterId) {
+ KubernetesClient kubernetesClient,
+ String namespace,
+ String clusterId,
+ long shutdownTimeout) {
boolean jobManagerRunning = true;
boolean serviceRunning = true;
- for (int i = 0; i < 60; i++) {
+ for (int i = 0; i < shutdownTimeout; i++) {
if (jobManagerRunning) {
PodList jmPodList = getJmPodList(kubernetesClient, namespace, clusterId);
@@ -207,7 +213,10 @@ public class FlinkUtils {
if (!jobManagerRunning && !serviceRunning) {
break;
}
- LOG.info("Waiting for cluster shutdown... ({})", i);
+ // log a message waiting to shutdown Flink cluster every 5 seconds.
+ if ((i + 1) % 5 == 0) {
+ LOG.info("Waiting for cluster shutdown... ({}s)", i + 1);
+ }
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
@@ -219,11 +228,12 @@ public class FlinkUtils {
/** Wait until the FLink cluster has completely shut down. */
public static void waitForClusterShutdown(
- KubernetesClient kubernetesClient, Configuration conf) {
+ KubernetesClient kubernetesClient, Configuration conf, long shutdownTimeout) {
FlinkUtils.waitForClusterShutdown(
kubernetesClient,
conf.getString(KubernetesConfigOptions.NAMESPACE),
- conf.getString(KubernetesConfigOptions.CLUSTER_ID));
+ conf.getString(KubernetesConfigOptions.CLUSTER_ID),
+ shutdownTimeout);
}
public static PodList getJmPodList(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index b8721e0..0764d9e 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -142,7 +142,10 @@ public class TestingFlinkService extends FlinkService {
@Override
public void stopSessionCluster(
- FlinkDeployment deployment, Configuration conf, boolean deleteHa) {
+ FlinkDeployment deployment,
+ Configuration conf,
+ boolean deleteHa,
+ long shutdownTimeout) {
sessions.remove(deployment.getMetadata().getName());
}
diff --git a/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml b/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml
index 6e6e104..a7254ff 100644
--- a/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml
+++ b/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml
@@ -18,9 +18,12 @@
# operator.reconciler.reschedule.interval: 60 s
# operator.reconciler.max.parallelism: 5
+# operator.reconciler.flink.cancel.job.timeout: 1min
+# operator.reconciler.flink.cluster.shutdown.timeout: 60s
# operator.observer.rest-ready.delay: 10 s
# operator.observer.progress-check.interval: 10 s
# operator.observer.savepoint.trigger.grace-period: 10 s
+# operator.observer.flink.client.timeout: 10 s
# metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
# metrics.reporter.slf4j.interval: 5 MINUTE