This documentation is for an unreleased version of the Apache Flink Kubernetes Operator. We recommend you use the latest stable version.

Configuration #

Specifying Operator Configuration #

The operator allows users to specify default configuration that will be shared by the Flink operator itself and the Flink deployments.

These configuration files are mounted externally via ConfigMaps. The Configuration files with default values are shipped in the Helm chart. It is recommended to review and adjust them if needed in the values.yaml file before deploying the Operator in production environments.

To append to the default configuration, simply define the flink-conf.yaml key in the defaultConfiguration section of the Helm values.yaml file:

  create: true
  # Set append to false to replace configuration files
  append: true
  flink-conf.yaml: |+
    # Flink Config Overrides
    kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
    kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE

    kubernetes.operator.reconcile.interval: 15 s 5 s

To learn more about metrics and logging configuration please refer to the dedicated docs page.

Dynamic Operator Configuration #

The Kubernetes operator supports dynamic config changes through the operator ConfigMaps. Dynamic operator configuration is enabled by default, and can be disabled by setting kubernetes.operator.dynamic.config.enabled to false. Time interval for checking dynamic config changes is specified by kubernetes.operator.dynamic.config.check.interval of which default value is 5 minutes.

Verify whether dynamic operator configuration updates is enabled via the deploy/flink-kubernetes-operator log has:

2022-05-28 13:08:29,222 o.a.f.k.o.c.FlinkConfigManager [INFO ] Enabled dynamic config updates, checking config changes every PT5M

To change config values dynamically the ConfigMap can be directly edited via kubectl patch or kubectl edit command. For example to change the reschedule interval you can override kubernetes.operator.reconcile.interval.

Verify whether the config value of kubernetes.operator.reconcile.interval is updated to 30 seconds via the deploy/flink-kubernetes-operator log has:

2022-05-28 13:08:30,115 o.a.f.k.o.c.FlinkConfigManager [INFO ] Updating default configuration to {kubernetes.operator.reconcile.interval=PT30S}

Operator Configuration Reference #

System Configuration #

General operator system configuration. Cannot be overridden on a per-resource basis.

Key Default Type Description
false Boolean Enables dynamic change of watched/monitored namespaces.
1 min Duration The timeout for the reconciler to wait for flink to cancel job.
10 s Duration The timeout for the observer to wait the flink rest client to return.
1 min Duration The interval for the controller to reschedule the reconcile process.
10 Integer The maximum number of threads running the reconciliation loop. Use -1 for infinite.
1 min Duration The timeout for the resource clean up to wait for flink to shutdown cluster.
5 s Duration Initial interval of automatic reconcile retries on recoverable errors.
2.0 Double Interval multiplier of automatic reconcile retries on recoverable errors.
10 Integer Max attempts of automatic reconcile retries on recoverable errors.
"/opt/flink/artifacts" String The base dir to put the session job artifacts.
"JOSDK_ALL_NAMESPACES" String Comma separated list of namespaces the operator monitors for custom resources.

Resource/User Configuration #

These options can be configured on both an operator and a per-resource level. When set under spec.flinkConfiguration for the Flink resources it will override the default value provided in the operator default configuration (flink-conf.yaml).

Key Default Type Description
1 min Duration The timeout for deployments to become ready/stable before being rolled back if rollback is enabled.
false Boolean Whether to enable rolling back failed deployment upgrades.
true Boolean Whether to enable recovery of missing/deleted jobmanager deployments.
false Boolean Whether to ignore pending savepoint during job upgrade.
true Boolean Enables last-state fallback for savepoint upgrade mode. When the job is not running thus savepoint cannot be triggered but HA metadata is available for last state restore the operator can initiate the upgrade process when the flag is enabled.
0 ms Duration Interval at which periodic savepoints will be triggered. The triggering schedule is not guaranteed, savepoints will be triggered as part of the regular reconcile loop.


Type of the binary format in which a savepoint should be taken.

Possible values:
  • "CANONICAL": A canonical, common for all state backends format. It lets you switch state backends.
  • "NATIVE": A format specific for the chosen state backend, in its native binary format. Might be faster to take and restore from than the canonical one.
86400000 ms Duration Maximum age for savepoint history entries to retain. Due to lazy clean-up, the most recent savepoint may live longer than the max age.
10 Integer Maximum number of savepoint history entries to retain.
1 min Duration The interval before a savepoint trigger attempt is marked as unsuccessful.
(none) Map Custom HTTP header for HttpArtifactFetcher. The header will be applied when getting the session job artifacts. Expected format: headerKey1:headerValue1,headerKey2:headerValue2.

System Metrics Configuration #

Operator system metrics configuration. Cannot be overridden on a per-resource basis.

Key Default Type Description
true Boolean Enable forwarding of Java Operator SDK metrics to the Flink metric registry.
true Boolean Enable Kubernetes Operator JVM metrics.
true Boolean Enable KubernetesClient metrics for measuring the HTTP traffic to the Kubernetes API Server.
1000 Integer Defines the number of measured samples when calculating statistics.
"<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>.<resourcetype>" String Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource.
"<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.<resourcetype>" String Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource namespace.
"<host>.k8soperator.<namespace>.<name>.system" String Defines the scope format string that is applied to all metrics scoped to the kubernetes operator.
true Boolean Enable resource lifecycle state metrics. This enables both state and transition counts/histograms.
true Boolean In addition to the system level histograms, enable per namespace tracking of state and transition times.
true Boolean Enables metrics for FlinkDeployment and FlinkSessionJob custom resources.

Advanced System Configuration #

Advanced operator system configuration. Cannot be overridden on a per-resource basis.

Key Default Type Description
1000 Integer Max config cache size.
10 min Duration Expiration time for cached configs.
5 min Duration Time interval for checking config changes.
true Boolean Whether to enable on-the-fly config changes through the operator configmap.
true Boolean Enables health probe for the kubernetes operator.
8085 Integer The port the health probe will use to expose the status.
(none) String Label selector of the custom resources to be watched. Please see for the format supported.
10 s Duration The interval for observing status for in-progress operations such as deployment and savepoints.
10 s Duration Final delay before deployment is marked ready after port becomes accessible.
(none) Duration Maximum age threshold for savepoint history entries to retain.
(none) Integer Maximum number threshold of savepoint history entries to retain.