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 5e3e9962 [FLINK-29313] Observe spec should include latest operator
configs + improved test coverage
5e3e9962 is described below
commit 5e3e9962906c605ea55d4caf2fef2ee5ba400019
Author: Gyula Fora <[email protected]>
AuthorDate: Mon Sep 19 15:58:42 2022 +0200
[FLINK-29313] Observe spec should include latest operator configs +
improved test coverage
---
.../kubernetes_operator_metric_configuration.html | 36 ++++-----
.../operator/config/FlinkConfigManager.java | 19 ++++-
.../config/KubernetesOperatorConfigOptions.java | 88 ++++++++++++----------
.../operator/crd/spec/AbstractFlinkSpec.java | 5 +-
.../operator/listener/ListenerUtils.java | 4 +-
.../metrics/KubernetesOperatorMetricOptions.java | 25 +++---
.../operator/metrics/OperatorMetricUtils.java | 4 +-
.../operator/observer/SavepointObserver.java | 30 +++++---
.../kubernetes/operator/TestingFlinkService.java | 23 +++---
.../operator/config/FlinkConfigManagerTest.java | 11 +++
.../controller/FlinkDeploymentControllerTest.java | 7 +-
.../operator/controller/RollbackTest.java | 26 +++++++
.../operator/observer/SavepointObserverTest.java | 10 +--
.../deployment/ApplicationObserverTest.java | 30 +++++++-
.../sessionjob/SessionJobObserverTest.java | 3 +-
.../deployment/ApplicationReconcilerTest.java | 27 +++++--
.../sessionjob/SessionJobReconcilerTest.java | 9 ++-
17 files changed, 236 insertions(+), 121 deletions(-)
diff --git
a/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
b/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
index 57859b88..939339e7 100644
---
a/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
+++
b/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
@@ -32,6 +32,24 @@
<td>Integer</td>
<td>Defines the number of measured samples when calculating
statistics.</td>
</tr>
+ <tr>
+
<td><h5>kubernetes.operator.metrics.scope.k8soperator.resource</h5></td>
+ <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>.<resourcetype>"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator resource.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.metrics.scope.k8soperator.resourcens</h5></td>
+ <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.<resourcetype>"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator resource namespace.</td>
+ </tr>
+ <tr>
+
<td><h5>kubernetes.operator.metrics.scope.k8soperator.system</h5></td>
+ <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.system"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.resource.lifecycle.metrics.enabled</h5></td>
<td style="word-wrap: break-word;">true</td>
@@ -50,23 +68,5 @@
<td>Boolean</td>
<td>Enables metrics for FlinkDeployment and FlinkSessionJob custom
resources.</td>
</tr>
- <tr>
- <td><h5>metrics.scope.k8soperator.resource</h5></td>
- <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>.<resourcetype>"</td>
- <td>String</td>
- <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator resource.</td>
- </tr>
- <tr>
- <td><h5>metrics.scope.k8soperator.resourcens</h5></td>
- <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.<resourcetype>"</td>
- <td>String</td>
- <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator resource namespace.</td>
- </tr>
- <tr>
- <td><h5>metrics.scope.k8soperator.system</h5></td>
- <td style="word-wrap:
break-word;">"<host>.k8soperator.<namespace>.<name>.system"</td>
- <td>String</td>
- <td>Defines the scope format string that is applied to all metrics
scoped to the kubernetes operator.</td>
- </tr>
</tbody>
</table>
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
index 0340891a..49181886 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.spec.AbstractFlinkSpec;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -52,6 +53,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_CONFIG_CHECK_INTERVAL;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_CONFIG_ENABLED;
@@ -157,7 +159,22 @@ public class FlinkConfigManager {
throw new RuntimeException(
"Cannot create observe config before first deployment,
this indicates a bug.");
}
- return getConfig(deployment.getMetadata(), deployedSpec);
+ var conf = getConfig(deployment.getMetadata(), deployedSpec);
+ addOperatorConfigsFromSpec(deployment.getSpec(), conf);
+ return conf;
+ }
+
+ private void addOperatorConfigsFromSpec(AbstractFlinkSpec spec,
Configuration conf) {
+ // Observe config should include the latest operator related settings
+ if (spec.getFlinkConfiguration() != null) {
+ spec.getFlinkConfiguration()
+ .forEach(
+ (k, v) -> {
+ if (k.startsWith(K8S_OP_CONF_PREFIX)) {
+ conf.setString(k, v);
+ }
+ });
+ }
}
public Configuration getSessionJobConfig(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index 5736c540..18664a1f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -25,6 +25,7 @@ import org.apache.flink.core.execution.SavepointFormatType;
import io.javaoperatorsdk.operator.api.config.ConfigurationService;
import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import org.jetbrains.annotations.NotNull;
import java.time.Duration;
import java.util.Map;
@@ -32,22 +33,33 @@ import java.util.Map;
/** This class holds configuration constants used by flink operator. */
public class KubernetesOperatorConfigOptions {
+ public static final String K8S_OP_CONF_PREFIX = "kubernetes.operator.";
+
public static final String SECTION_SYSTEM = "system";
public static final String SECTION_ADVANCED = "system_advanced";
public static final String SECTION_DYNAMIC = "dynamic";
+ @NotNull
+ public static ConfigOptions.OptionBuilder operatorConfig(String key) {
+ return ConfigOptions.key(K8S_OP_CONF_PREFIX + key);
+ }
+
+ public static String operatorConfigKey(String key) {
+ return K8S_OP_CONF_PREFIX + key;
+ }
+
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_RECONCILE_INTERVAL =
- ConfigOptions.key("kubernetes.operator.reconcile.interval")
+ operatorConfig("reconcile.interval")
.durationType()
.defaultValue(Duration.ofSeconds(60))
-
.withDeprecatedKeys("kubernetes.operator.reconciler.reschedule.interval")
+ .withDeprecatedKeys(K8S_OP_CONF_PREFIX +
"reconciler.reschedule.interval")
.withDescription(
"The interval for the controller to reschedule the
reconcile process.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration>
OPERATOR_OBSERVER_REST_READY_DELAY =
- ConfigOptions.key("kubernetes.operator.observer.rest-ready.delay")
+ operatorConfig("observer.rest-ready.delay")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
@@ -55,16 +67,16 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Integer> OPERATOR_RECONCILE_PARALLELISM =
- ConfigOptions.key("kubernetes.operator.reconcile.parallelism")
+ operatorConfig("reconcile.parallelism")
.intType()
.defaultValue(ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER)
-
.withDeprecatedKeys("kubernetes.operator.reconciler.max.parallelism")
+
.withDeprecatedKeys(operatorConfigKey("reconciler.max.parallelism"))
.withDescription(
"The maximum number of threads running the
reconciliation loop. Use -1 for infinite.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration>
OPERATOR_OBSERVER_PROGRESS_CHECK_INTERVAL =
-
ConfigOptions.key("kubernetes.operator.observer.progress-check.interval")
+ operatorConfig("observer.progress-check.interval")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
@@ -72,52 +84,52 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration>
OPERATOR_SAVEPOINT_TRIGGER_GRACE_PERIOD =
-
ConfigOptions.key("kubernetes.operator.savepoint.trigger.grace-period")
+ operatorConfig("savepoint.trigger.grace-period")
.durationType()
.defaultValue(Duration.ofMinutes(1))
.withDeprecatedKeys(
-
"kubernetes.operator.observer.savepoint.trigger.grace-period")
+
operatorConfigKey("observer.savepoint.trigger.grace-period"))
.withDescription(
"The interval before a savepoint trigger attempt
is marked as unsuccessful.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_FLINK_CLIENT_TIMEOUT =
- ConfigOptions.key("kubernetes.operator.flink.client.timeout")
+ operatorConfig("flink.client.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(10))
-
.withDeprecatedKeys("kubernetes.operator.observer.flink.client.timeout")
+
.withDeprecatedKeys(operatorConfigKey("observer.flink.client.timeout"))
.withDescription(
"The timeout for the observer to wait the flink
rest client to return.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration>
OPERATOR_FLINK_CLIENT_CANCEL_TIMEOUT =
-
ConfigOptions.key("kubernetes.operator.flink.client.cancel.timeout")
+ operatorConfig("flink.client.cancel.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(1))
-
.withDeprecatedKeys("kubernetes.operator.reconciler.flink.cancel.job.timeout")
+
.withDeprecatedKeys(operatorConfigKey("reconciler.flink.cancel.job.timeout"))
.withDescription(
"The timeout for the reconciler to wait for flink
to cancel job.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration>
OPERATOR_RESOURCE_CLEANUP_TIMEOUT =
- ConfigOptions.key("kubernetes.operator.resource.cleanup.timeout")
+ operatorConfig("resource.cleanup.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDeprecatedKeys(
-
"kubernetes.operator.reconciler.flink.cluster.shutdown.timeout")
+
operatorConfigKey("reconciler.flink.cluster.shutdown.timeout"))
.withDescription(
"The timeout for the resource clean up to wait for
flink to shutdown cluster.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> DEPLOYMENT_ROLLBACK_ENABLED =
-
ConfigOptions.key("kubernetes.operator.deployment.rollback.enabled")
+ operatorConfig("deployment.rollback.enabled")
.booleanType()
.defaultValue(false)
.withDescription("Whether to enable rolling back failed
deployment upgrades.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> DEPLOYMENT_READINESS_TIMEOUT =
-
ConfigOptions.key("kubernetes.operator.deployment.readiness.timeout")
+ operatorConfig("deployment.readiness.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(1))
.withDescription(
@@ -126,21 +138,21 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<String> OPERATOR_USER_ARTIFACTS_BASE_DIR =
- ConfigOptions.key("kubernetes.operator.user.artifacts.base.dir")
+ operatorConfig("user.artifacts.base.dir")
.stringType()
.defaultValue("/opt/flink/artifacts")
.withDescription("The base dir to put the session job
artifacts.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean>
JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT =
-
ConfigOptions.key("kubernetes.operator.job.upgrade.ignore-pending-savepoint")
+ operatorConfig("job.upgrade.ignore-pending-savepoint")
.booleanType()
.defaultValue(false)
.withDescription("Whether to ignore pending savepoint
during job upgrade.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Boolean> OPERATOR_DYNAMIC_CONFIG_ENABLED =
- ConfigOptions.key("kubernetes.operator.dynamic.config.enabled")
+ operatorConfig("dynamic.config.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
@@ -148,38 +160,38 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration>
OPERATOR_DYNAMIC_CONFIG_CHECK_INTERVAL =
-
ConfigOptions.key("kubernetes.operator.dynamic.config.check.interval")
+ operatorConfig("dynamic.config.check.interval")
.durationType()
.defaultValue(Duration.ofMinutes(5))
.withDescription("Time interval for checking config
changes.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_CONFIG_CACHE_TIMEOUT =
- ConfigOptions.key("kubernetes.operator.config.cache.timeout")
+ operatorConfig("config.cache.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(10))
.withDescription("Expiration time for cached configs.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Integer> OPERATOR_CONFIG_CACHE_SIZE =
- ConfigOptions.key("kubernetes.operator.config.cache.size")
+ operatorConfig("config.cache.size")
.intType()
.defaultValue(1000)
.withDescription("Max config cache size.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean>
OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED =
-
ConfigOptions.key("kubernetes.operator.jm-deployment-recovery.enabled")
+ operatorConfig("jm-deployment-recovery.enabled")
.booleanType()
.defaultValue(true)
.withDeprecatedKeys(
-
"kubernetes.operator.reconciler.jm-deployment-recovery.enabled")
+
operatorConfigKey("reconciler.jm-deployment-recovery.enabled"))
.withDescription(
"Whether to enable recovery of missing/deleted
jobmanager deployments.");
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Integer>
OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT =
-
ConfigOptions.key("kubernetes.operator.savepoint.history.max.count")
+ operatorConfig("savepoint.history.max.count")
.intType()
.defaultValue(10)
.withDescription("Maximum number of savepoint history
entries to retain.");
@@ -194,7 +206,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration>
OPERATOR_SAVEPOINT_HISTORY_MAX_AGE =
- ConfigOptions.key("kubernetes.operator.savepoint.history.max.age")
+ operatorConfig("savepoint.history.max.age")
.durationType()
.defaultValue(Duration.ofHours(24))
.withDescription(
@@ -210,7 +222,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Map<String, String>>
JAR_ARTIFACT_HTTP_HEADER =
- ConfigOptions.key("kubernetes.operator.user.artifacts.http.header")
+ operatorConfig("user.artifacts.http.header")
.mapType()
.noDefaultValue()
.withDescription(
@@ -219,7 +231,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> PERIODIC_SAVEPOINT_INTERVAL =
-
ConfigOptions.key("kubernetes.operator.periodic.savepoint.interval")
+ operatorConfig("periodic.savepoint.interval")
.durationType()
.defaultValue(Duration.ZERO)
.withDescription(
@@ -228,7 +240,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<String> OPERATOR_WATCHED_NAMESPACES =
- ConfigOptions.key("kubernetes.operator.watched.namespaces")
+ operatorConfig("watched.namespaces")
.stringType()
.defaultValue(Constants.WATCH_ALL_NAMESPACES)
.withDescription(
@@ -236,7 +248,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<String> OPERATOR_LABEL_SELECTOR =
- ConfigOptions.key("kubernetes.operator.label.selector")
+ operatorConfig("label.selector")
.stringType()
.noDefaultValue()
.withDescription(
@@ -245,14 +257,14 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Boolean>
OPERATOR_DYNAMIC_NAMESPACES_ENABLED =
- ConfigOptions.key("kubernetes.operator.dynamic.namespaces.enabled")
+ operatorConfig("dynamic.namespaces.enabled")
.booleanType()
.defaultValue(false)
.withDescription("Enables dynamic change of
watched/monitored namespaces.");
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_RETRY_INITIAL_INTERVAL
=
- ConfigOptions.key("kubernetes.operator.retry.initial.interval")
+ operatorConfig("retry.initial.interval")
.durationType()
.defaultValue(Duration.ofSeconds(5))
.withDescription(
@@ -260,7 +272,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Double>
OPERATOR_RETRY_INTERVAL_MULTIPLIER =
- ConfigOptions.key("kubernetes.operator.retry.interval.multiplier")
+ operatorConfig("retry.interval.multiplier")
.doubleType()
.defaultValue(2.0)
.withDescription(
@@ -268,7 +280,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Integer> OPERATOR_RETRY_MAX_ATTEMPTS =
- ConfigOptions.key("kubernetes.operator.retry.max.attempts")
+ operatorConfig("retry.max.attempts")
.intType()
.defaultValue(10)
.withDescription(
@@ -276,7 +288,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean>
OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED =
-
ConfigOptions.key("kubernetes.operator.job.upgrade.last-state-fallback.enabled")
+ operatorConfig("job.upgrade.last-state-fallback.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
@@ -284,7 +296,7 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<SavepointFormatType>
OPERATOR_SAVEPOINT_FORMAT_TYPE =
- ConfigOptions.key("kubernetes.operator.savepoint.format.type")
+ operatorConfig("savepoint.format.type")
.enumType(SavepointFormatType.class)
.defaultValue(SavepointFormatType.DEFAULT)
.withDescription(
@@ -292,14 +304,14 @@ public class KubernetesOperatorConfigOptions {
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Boolean> OPERATOR_HEALTH_PROBE_ENABLED =
- ConfigOptions.key("kubernetes.operator.health.probe.enabled")
+ operatorConfig("health.probe.enabled")
.booleanType()
.defaultValue(true)
.withDescription("Enables health probe for the kubernetes
operator.");
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Integer> OPERATOR_HEALTH_PROBE_PORT =
- ConfigOptions.key("kubernetes.operator.health.probe.port")
+ operatorConfig("health.probe.port")
.intType()
.defaultValue(8085)
.withDescription("The port the health probe will use to
expose the status.");
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java
index d7d1e2a5..b5e3a1fd 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java
@@ -31,6 +31,8 @@ import lombok.experimental.SuperBuilder;
import java.util.Map;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
+
/** The common spec. */
@Experimental
@Data
@@ -51,8 +53,7 @@ public abstract class AbstractFlinkSpec implements
Diffable<AbstractFlinkSpec> {
/** Flink configuration overrides for the Flink deployment or Flink
session job. */
@SpecDiff.Config({
@SpecDiff.Entry(prefix = "parallelism.default", type =
DiffType.IGNORE),
- @SpecDiff.Entry(prefix = "kubernetes.operator", type =
DiffType.IGNORE),
- @SpecDiff.Entry(prefix = "metrics.scope.k8soperator", type =
DiffType.IGNORE)
+ @SpecDiff.Entry(prefix = K8S_OP_CONF_PREFIX, type = DiffType.IGNORE),
})
private Map<String, String> flinkConfiguration;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/ListenerUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/ListenerUtils.java
index 843cdf26..d99fd028 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/ListenerUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/ListenerUtils.java
@@ -41,12 +41,14 @@ import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
+
/** Flink resource listener utilities. */
public class ListenerUtils {
private static final Logger LOG =
LoggerFactory.getLogger(ListenerUtils.class);
- private static final String PREFIX =
"kubernetes.operator.plugins.listeners.";
+ private static final String PREFIX = K8S_OP_CONF_PREFIX +
"plugins.listeners.";
private static final String SUFFIX = ".class";
private static final Pattern PTN =
Pattern.compile(Pattern.quote(PREFIX) + "([\\S&&[^.]]*)" +
Pattern.quote(SUFFIX));
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
index 9da0b360..dfc876d4 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
@@ -18,61 +18,62 @@
package org.apache.flink.kubernetes.operator.metrics;
import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
+
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.operatorConfig;
/** Configuration options for metrics. */
public class KubernetesOperatorMetricOptions {
public static final ConfigOption<Boolean> OPERATOR_JVM_METRICS_ENABLED =
- ConfigOptions.key("kubernetes.operator.jvm.metrics.enabled")
+ operatorConfig("jvm.metrics.enabled")
.booleanType()
.defaultValue(true)
.withDescription("Enable Kubernetes Operator JVM
metrics.");
public static final ConfigOption<Boolean> OPERATOR_JOSDK_METRICS_ENABLED =
- ConfigOptions.key("kubernetes.operator.josdk.metrics.enabled")
+ operatorConfig("josdk.metrics.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
"Enable forwarding of Java Operator SDK metrics to
the Flink metric registry.");
public static final ConfigOption<Boolean>
OPERATOR_KUBERNETES_CLIENT_METRICS_ENABLED =
-
ConfigOptions.key("kubernetes.operator.kubernetes.client.metrics.enabled")
+ operatorConfig("kubernetes.client.metrics.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
"Enable KubernetesClient metrics for measuring the
HTTP traffic to the Kubernetes API Server.");
public static final ConfigOption<Boolean>
OPERATOR_RESOURCE_METRICS_ENABLED =
- ConfigOptions.key("kubernetes.operator.resource.metrics.enabled")
+ operatorConfig("resource.metrics.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
"Enables metrics for FlinkDeployment and
FlinkSessionJob custom resources.");
public static final ConfigOption<Boolean>
OPERATOR_LIFECYCLE_METRICS_ENABLED =
-
ConfigOptions.key("kubernetes.operator.resource.lifecycle.metrics.enabled")
+ operatorConfig("resource.lifecycle.metrics.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
"Enable resource lifecycle state metrics. This
enables both state and transition counts/histograms.");
public static final ConfigOption<Boolean>
OPERATOR_LIFECYCLE_NAMESPACE_HISTOGRAMS_ENABLED =
-
ConfigOptions.key("kubernetes.operator.resource.lifecycle.namespace.histograms.enabled")
+ operatorConfig("resource.lifecycle.namespace.histograms.enabled")
.booleanType()
.defaultValue(true)
.withDescription(
"In addition to the system level histograms,
enable per namespace tracking of state and transition times.");
public static final ConfigOption<Integer>
OPERATOR_METRICS_HISTOGRAM_SAMPLE_SIZE =
-
ConfigOptions.key("kubernetes.operator.metrics.histogram.sample.size")
+ operatorConfig("metrics.histogram.sample.size")
.intType()
.defaultValue(1000)
.withDescription(
"Defines the number of measured samples when
calculating statistics.");
public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR =
- ConfigOptions.key("metrics.scope.k8soperator.system")
+ operatorConfig("metrics.scope.k8soperator.system")
.stringType()
.defaultValue("<host>.k8soperator.<namespace>.<name>.system")
.withDeprecatedKeys("metrics.scope.k8soperator")
@@ -80,18 +81,20 @@ public class KubernetesOperatorMetricOptions {
"Defines the scope format string that is applied
to all metrics scoped to the kubernetes operator.");
public static final ConfigOption<String>
SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS =
- ConfigOptions.key("metrics.scope.k8soperator.resourcens")
+ operatorConfig("metrics.scope.k8soperator.resourcens")
.stringType()
.defaultValue(
"<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.<resourcetype>")
+ .withDeprecatedKeys("metrics.scope.k8soperator.resourcens")
.withDescription(
"Defines the scope format string that is applied
to all metrics scoped to the kubernetes operator resource namespace.");
public static final ConfigOption<String>
SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE =
- ConfigOptions.key("metrics.scope.k8soperator.resource")
+ operatorConfig("metrics.scope.k8soperator.resource")
.stringType()
.defaultValue(
"<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>.<resourcetype>")
+ .withDeprecatedKeys("metrics.scope.k8soperator.resource")
.withDescription(
"Defines the scope format string that is applied
to all metrics scoped to the kubernetes operator resource.");
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
index c5f6be75..155817a8 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
@@ -43,12 +43,14 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
+
/** Utility class for flink based operator metrics. */
public class OperatorMetricUtils {
private static final Logger LOG =
LoggerFactory.getLogger(OperatorMetricUtils.class);
- private static final String OPERATOR_METRICS_PREFIX =
"kubernetes.operator.metrics.";
+ private static final String OPERATOR_METRICS_PREFIX = K8S_OP_CONF_PREFIX +
"metrics.";
private static final String METRICS_PREFIX = "metrics.";
public static KubernetesOperatorMetricGroup
initOperatorMetrics(Configuration defaultConfig) {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
index 091badfc..aa2d69de 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
@@ -74,6 +74,8 @@ public class SavepointObserver<
if (ReconciliationUtils.isJobInTerminalState(resource.getStatus())) {
observeLatestSavepoint(savepointInfo, jobId, deployedConfig);
}
+
+ cleanupSavepointHistory(savepointInfo, deployedConfig);
}
/**
@@ -129,25 +131,29 @@ public class SavepointObserver<
ReconciliationUtils.updateLastReconciledSavepointTriggerNonce(savepointInfo,
resource);
savepointInfo.updateLastSavepoint(savepoint);
- cleanupSavepointHistory(savepointInfo, savepoint, deployedConfig);
}
/** Clean up and dispose savepoints according to the configured max
size/age. */
@VisibleForTesting
- void cleanupSavepointHistory(
- SavepointInfo currentSavepointInfo,
- Savepoint newSavepoint,
- Configuration deployedConfig) {
+ void cleanupSavepointHistory(SavepointInfo currentSavepointInfo,
Configuration deployedConfig) {
// maintain history
List<Savepoint> savepointHistory =
currentSavepointInfo.getSavepointHistory();
+ if (savepointHistory.size() < 2) {
+ return;
+ }
+ var lastSavepoint = savepointHistory.get(savepointHistory.size() - 1);
+
int maxCount =
- ConfigOptionUtils.getValueWithThreshold(
- deployedConfig,
-
KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT,
- configManager
- .getOperatorConfiguration()
- .getSavepointHistoryCountThreshold());
+ Math.max(
+ 1,
+ ConfigOptionUtils.getValueWithThreshold(
+ deployedConfig,
+ KubernetesOperatorConfigOptions
+ .OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT,
+ configManager
+ .getOperatorConfiguration()
+ .getSavepointHistoryCountThreshold()));
while (savepointHistory.size() > maxCount) {
// remove oldest entries
disposeSavepointQuietly(savepointHistory.remove(0),
deployedConfig);
@@ -162,7 +168,7 @@ public class SavepointObserver<
Iterator<Savepoint> it = savepointHistory.iterator();
while (it.hasNext()) {
Savepoint sp = it.next();
- if (sp.getTimeStamp() < maxTms && sp != newSavepoint) {
+ if (sp.getTimeStamp() < maxTms && sp != lastSavepoint) {
it.remove();
disposeSavepointQuietly(sp, deployedConfig);
}
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 f25118e1..c030db69 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
@@ -19,7 +19,7 @@ package org.apache.flink.kubernetes.operator;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptions;
@@ -89,7 +89,7 @@ public class TestingFlinkService extends AbstractFlinkService
{
private int savepointCounter = 0;
private int triggerCounter = 0;
- private final List<Tuple2<String, JobStatusMessage>> jobs = new
ArrayList<>();
+ private final List<Tuple3<String, JobStatusMessage, Configuration>> jobs =
new ArrayList<>();
private final Map<JobID, String> jobErrors = new HashMap<>();
private final Set<String> sessions = new HashSet<>();
private boolean isPortReady = true;
@@ -98,8 +98,8 @@ public class TestingFlinkService extends AbstractFlinkService
{
private Runnable sessionJobSubmittedCallback;
private PodList podList = new PodList();
private Consumer<Configuration> listJobConsumer = conf -> {};
- private List<String> disposedSavepoints = new ArrayList<>();
- private Map<String, Boolean> savepointTriggers = new HashMap<>();
+ private final List<String> disposedSavepoints = new ArrayList<>();
+ private final Map<String, Boolean> savepointTriggers = new HashMap<>();
private int desiredReplicas = 0;
public TestingFlinkService() {
@@ -157,7 +157,8 @@ public class TestingFlinkService extends
AbstractFlinkService {
JobStatus.RUNNING,
System.currentTimeMillis());
- jobs.add(Tuple2.of(conf.get(SavepointConfigOptions.SAVEPOINT_PATH),
jobStatusMessage));
+ jobs.add(
+ Tuple3.of(conf.get(SavepointConfigOptions.SAVEPOINT_PATH),
jobStatusMessage, conf));
}
protected void validateHaMetadataExists(Configuration conf) {
@@ -214,7 +215,7 @@ public class TestingFlinkService extends
AbstractFlinkService {
JobStatus.RUNNING,
System.currentTimeMillis());
- jobs.add(Tuple2.of(savepoint, jobStatusMessage));
+ jobs.add(Tuple3.of(savepoint, jobStatusMessage, conf));
if (sessionJobSubmittedCallback != null) {
sessionJobSubmittedCallback.run();
}
@@ -233,7 +234,7 @@ public class TestingFlinkService extends
AbstractFlinkService {
this.listJobConsumer = listJobConsumer;
}
- public List<Tuple2<String, JobStatusMessage>> listJobs() {
+ public List<Tuple3<String, JobStatusMessage, Configuration>> listJobs() {
return jobs;
}
@@ -354,8 +355,7 @@ public class TestingFlinkService extends
AbstractFlinkService {
private String cancelJob(FlinkVersion flinkVersion, JobID jobID, boolean
savepoint)
throws Exception {
- Optional<Tuple2<String, JobStatusMessage>> jobOpt =
- jobs.stream().filter(js ->
js.f1.getJobId().equals(jobID)).findAny();
+ var jobOpt = jobs.stream().filter(js ->
js.f1.getJobId().equals(jobID)).findAny();
if (jobOpt.isEmpty()) {
throw new Exception("Job not found");
@@ -402,14 +402,13 @@ public class TestingFlinkService extends
AbstractFlinkService {
@Override
public Optional<Savepoint> getLastCheckpoint(JobID jobId, Configuration
conf) throws Exception {
- Optional<Tuple2<String, JobStatusMessage>> jobOpt =
- jobs.stream().filter(js ->
js.f1.getJobId().equals(jobId)).findAny();
+ var jobOpt = jobs.stream().filter(js ->
js.f1.getJobId().equals(jobId)).findAny();
if (jobOpt.isEmpty()) {
throw new Exception("Job not found");
}
- Tuple2<String, JobStatusMessage> t = jobOpt.get();
+ var t = jobOpt.get();
if (!t.f1.getJobState().isGloballyTerminalState()) {
throw new Exception("Checkpoint should not be queried if job is
not in terminal state");
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
index abeee8b3..bdd101b0 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
@@ -54,6 +54,8 @@ public class FlinkConfigManagerTest {
@Test
public void testConfigGeneration() {
ConfigOption<String> testConf =
ConfigOptions.key("test").stringType().noDefaultValue();
+ ConfigOption<String> opTestConf =
+
ConfigOptions.key("kubernetes.operator.test").stringType().noDefaultValue();
FlinkConfigManager configManager =
new FlinkConfigManager(
@@ -67,16 +69,25 @@ public class FlinkConfigManagerTest {
var reconciliationStatus =
deployment.getStatus().getReconciliationStatus();
deployment.getSpec().getFlinkConfiguration().put(testConf.key(),
"reconciled");
+ deployment.getSpec().getFlinkConfiguration().put(opTestConf.key(),
"reconciled");
reconciliationStatus.serializeAndSetLastReconciledSpec(deployment.getSpec(),
deployment);
reconciliationStatus.markReconciledSpecAsStable();
deployment.getSpec().getFlinkConfiguration().put(testConf.key(),
"latest");
+ deployment.getSpec().getFlinkConfiguration().put(opTestConf.key(),
"latest");
+
assertEquals(
"latest",
configManager
.getDeployConfig(deployment.getMetadata(),
deployment.getSpec())
.get(testConf));
+ assertEquals(
+ "latest",
+ configManager
+ .getDeployConfig(deployment.getMetadata(),
deployment.getSpec())
+ .get(opTestConf));
assertEquals("reconciled",
configManager.getObserveConfig(deployment).get(testConf));
+ assertEquals("latest",
configManager.getObserveConfig(deployment).get(opTestConf));
deployment.getSpec().getFlinkConfiguration().put(testConf.key(),
"stable");
reconciliationStatus.serializeAndSetLastReconciledSpec(deployment.getSpec(),
deployment);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index e0fafc47..f5bd8d4c 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -17,7 +17,6 @@
package org.apache.flink.kubernetes.operator.controller;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
@@ -381,7 +380,7 @@ public class FlinkDeploymentControllerTest {
CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
"file:///flink-data/savepoints");
testController.reconcile(appCluster, context);
- List<Tuple2<String, JobStatusMessage>> jobs = flinkService.listJobs();
+ var jobs = flinkService.listJobs();
assertEquals(1, jobs.size());
assertEquals("s0", jobs.get(0).f0);
assertEquals(
@@ -389,7 +388,7 @@ public class FlinkDeploymentControllerTest {
"component=taskmanager,app=" +
appCluster.getMetadata().getName(), 1),
appCluster.getStatus().getTaskManager());
- List<Tuple2<String, JobStatusMessage>> previousJobs = new
ArrayList<>(jobs);
+ var previousJobs = new ArrayList<>(jobs);
appCluster.getSpec().getJob().setInitialSavepointPath("s1");
// Send in a no-op change
@@ -470,7 +469,7 @@ public class FlinkDeploymentControllerTest {
appCluster.getSpec().getJob().setInitialSavepointPath("s0");
testController.reconcile(appCluster, context);
- List<Tuple2<String, JobStatusMessage>> jobs = flinkService.listJobs();
+ var jobs = flinkService.listJobs();
assertEquals(1, jobs.size());
assertEquals("s0", jobs.get(0).f0);
assertEquals(1, testController.events().size());
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index a6a57b4b..0e34d05a 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -156,11 +156,13 @@ public class RollbackTest {
public void testRollbackFailureWithLastState(FlinkVersion flinkVersion)
throws Exception {
var dep = TestUtils.buildApplicationCluster(flinkVersion);
dep.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ dep.getSpec().getFlinkConfiguration().put("t", "1");
testRollback(
dep,
() -> {
dep.getSpec().getJob().setParallelism(9999);
+ dep.getSpec().getFlinkConfiguration().remove("t");
testController.reconcile(dep, context);
assertEquals(
JobState.SUSPENDED,
@@ -179,6 +181,17 @@ public class RollbackTest {
assertEquals("RUNNING",
dep.getStatus().getJobStatus().getState());
assertEquals(1, flinkService.listJobs().size());
+ // Trigger deployment recovery
+ flinkService.clear();
+ flinkService.setPortReady(false);
+
+ testController.reconcile(dep, context);
+ flinkService.setPortReady(true);
+ testController.reconcile(dep, context);
+ var jobs = flinkService.listJobs();
+ // Make sure deployment was recovered with correct
spec/config
+ assertTrue(jobs.get(jobs.size() - 1).f2.containsKey("t"));
+
// Remove job to simulate rollback failure
flinkService.clear();
flinkService.setPortReady(false);
@@ -199,6 +212,12 @@ public class RollbackTest {
testRollback(
dep,
() -> {
+ dep.getSpec()
+ .getFlinkConfiguration()
+ .put(
+
KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED
+ .key(),
+ "false");
dep.getSpec().getJob().setParallelism(9999);
testController.reconcile(dep, context);
assertEquals(
@@ -209,6 +228,13 @@ public class RollbackTest {
.getJob()
.getState());
testController.reconcile(dep, context);
+ // Validate that rollback config is picked up from latest
deploy conf
+ dep.getSpec()
+ .getFlinkConfiguration()
+ .put(
+
KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED
+ .key(),
+ "true");
// Trigger rollback by delaying the recovery
Thread.sleep(200);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
index 6a94fe3c..af42289f 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
@@ -65,7 +65,7 @@ public class SavepointObserverTest {
Savepoint sp = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
spInfo.updateLastSavepoint(sp);
- observer.cleanupSavepointHistory(spInfo, sp,
configManager.getDefaultConfig());
+ observer.cleanupSavepointHistory(spInfo,
configManager.getDefaultConfig());
Assertions.assertNotNull(spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
@@ -83,7 +83,7 @@ public class SavepointObserverTest {
Savepoint sp1 = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
spInfo.updateLastSavepoint(sp1);
- observer.cleanupSavepointHistory(spInfo, sp1, conf);
+ observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp1), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
@@ -91,7 +91,7 @@ public class SavepointObserverTest {
Savepoint sp2 = new Savepoint(2, "sp2", SavepointTriggerType.MANUAL);
spInfo.updateLastSavepoint(sp2);
- observer.cleanupSavepointHistory(spInfo, sp2, conf);
+ observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp2), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
@@ -112,7 +112,7 @@ public class SavepointObserverTest {
Savepoint sp1 = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
spInfo.updateLastSavepoint(sp1);
- observer.cleanupSavepointHistory(spInfo, sp1, conf);
+ observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp1), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
@@ -120,7 +120,7 @@ public class SavepointObserverTest {
Savepoint sp2 = new Savepoint(2, "sp2", SavepointTriggerType.MANUAL);
spInfo.updateLastSavepoint(sp2);
- observer.cleanupSavepointHistory(spInfo, sp2, conf);
+ observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp2), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index aa9504de..cbe59f85 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -18,11 +18,11 @@
package org.apache.flink.kubernetes.operator.observer.deployment;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
@@ -379,7 +379,7 @@ public class ApplicationObserverTest {
flinkService.setPortReady(true);
observer.observe(deployment, readyContext);
// Simulate Failed job
- Tuple2<String, JobStatusMessage> jobTuple =
flinkService.listJobs().get(0);
+ var jobTuple = flinkService.listJobs().get(0);
jobTuple.f0 = "last-SP";
jobTuple.f1 =
new JobStatusMessage(
@@ -403,6 +403,32 @@ public class ApplicationObserverTest {
.getLastSavepoint()
.getLocation());
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
+
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ 3,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .size());
+
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(
+
KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT.key(),
+ "1");
+ observer.observe(deployment, readyContext);
+ assertEquals(
+ 1,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .size());
}
private void bringToReadyStatus(FlinkDeployment deployment) {
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
index 37db442b..05ad76f3 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
@@ -19,7 +19,6 @@ package
org.apache.flink.kubernetes.operator.observer.sessionjob;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
@@ -326,7 +325,7 @@ public class SessionJobObserverTest {
// mock a job with different id of the target CR occurs
var jobs = flinkService.listJobs();
- for (Tuple2<String, JobStatusMessage> job : jobs) {
+ for (var job : jobs) {
if (!job.f1.getJobState().isGloballyTerminalState()
&& !job.f1.getJobId().toHexString().equals(jobID)) {
job.f1 =
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index e1f04ec5..b8f5e9af 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -17,7 +17,7 @@
package org.apache.flink.kubernetes.operator.reconciler.deployment;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
@@ -100,7 +100,7 @@ public class ApplicationReconcilerTest {
FlinkDeployment deployment =
TestUtils.buildApplicationCluster(flinkVersion);
reconciler.reconcile(deployment, context);
- List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
// Test stateless upgrade
@@ -228,7 +228,7 @@ public class ApplicationReconcilerTest {
final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
- List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
// Suspend FlinkDeployment with savepoint upgrade mode
@@ -266,7 +266,7 @@ public class ApplicationReconcilerTest {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
- List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
@@ -346,6 +346,16 @@ public class ApplicationReconcilerTest {
spDeployment.getSpec().getJob().setSavepointTriggerNonce(null);
reconciler.reconcile(spDeployment, context);
assertFalse(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
+
+ reconciler.reconcile(spDeployment, context);
+
assertFalse(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
+
+ spDeployment
+ .getSpec()
+ .getFlinkConfiguration()
+
.put(KubernetesOperatorConfigOptions.PERIODIC_SAVEPOINT_INTERVAL.key(), "1");
+ reconciler.reconcile(spDeployment, context);
+
assertTrue(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
}
@Test
@@ -389,7 +399,7 @@ public class ApplicationReconcilerTest {
.getImage());
// Upgrade mode changes from stateless to last-state should trigger a
savepoint
final String expectedSavepointPath = "savepoint_0";
- final List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
assertEquals(expectedSavepointPath, runningJobs.get(0).f0);
}
@@ -431,7 +441,7 @@ public class ApplicationReconcilerTest {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
- List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
// Test restart job
@@ -468,7 +478,8 @@ public class ApplicationReconcilerTest {
}
private void verifyAndSetRunningJobsToStatus(
- FlinkDeployment deployment, List<Tuple2<String, JobStatusMessage>>
runningJobs) {
+ FlinkDeployment deployment,
+ List<Tuple3<String, JobStatusMessage, Configuration>> runningJobs)
{
assertEquals(1, runningJobs.size());
assertNull(runningJobs.get(0).f0);
deployment
@@ -487,7 +498,7 @@ public class ApplicationReconcilerTest {
public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
reconciler.reconcile(deployment, context);
- List<Tuple2<String, JobStatusMessage>> runningJobs =
flinkService.listJobs();
+ var runningJobs = flinkService.listJobs();
verifyAndSetRunningJobsToStatus(deployment, runningJobs);
FlinkDeployment spDeployment = ReconciliationUtils.clone(deployment);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
index 6b1cf078..13ef730a 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.kubernetes.operator.reconciler.sessionjob;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
@@ -394,8 +394,9 @@ public class SessionJobReconcilerTest {
assertFalse(SavepointUtils.savepointInProgress(sp1SessionJob.getStatus().getJobStatus()));
}
- private Tuple2<String, JobStatusMessage> verifyAndReturnTheSubmittedJob(
- FlinkSessionJob sessionJob, List<Tuple2<String, JobStatusMessage>>
jobs) {
+ private Tuple3<String, JobStatusMessage, Configuration>
verifyAndReturnTheSubmittedJob(
+ FlinkSessionJob sessionJob,
+ List<Tuple3<String, JobStatusMessage, Configuration>> jobs) {
var jobID =
JobID.fromHexString(sessionJob.getStatus().getJobStatus().getJobId());
var submittedJobInfo =
jobs.stream().filter(t ->
t.f1.getJobId().equals(jobID)).findAny().get();
@@ -408,7 +409,7 @@ public class SessionJobReconcilerTest {
JobState expectedState,
String jobStatusObserved,
@Nullable String expectedSavepointPath,
- List<Tuple2<String, JobStatusMessage>> jobs) {
+ List<Tuple3<String, JobStatusMessage, Configuration>> jobs) {
var submittedJobInfo = verifyAndReturnTheSubmittedJob(sessionJob,
jobs);
assertEquals(expectedSavepointPath, submittedJobInfo.f0);
