You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "junzhong qin (Jira)" <ji...@apache.org> on 2026/02/06 11:00:00 UTC
[jira] [Updated] (FLINK-39039) Expose JOSDK cacheSyncTimeout as operator config
[ https://issues.apache.org/jira/browse/FLINK-39039?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
junzhong qin updated FLINK-39039:
---------------------------------
Description:
There are {color:#000000}thousand{color}s Flink deployments in our K8s cluster. When restart the operator, we found the Informer startup error, the log is as follows:
{code:java}
// Operator start log
2026-02-06 10:56:30,688 INFO io.javaoperatorsdk.operator.Operator [] [.] - Operator SDK 4.9.4 (commit: 3588780) built on 2024-08-30T00:52:57.000+0800 starting...2026-02-06 10:56:30,689 INFO io.javaoperatorsdk.operator.Operator [] [.] - Client version: 6.13.22026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkdeploymentcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController, resource: org.apache.flink.kubernetes.operator.api.FlinkDeployment2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkstatesnapshotcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotController, resource: org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinksessionjobcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController, resource: org.apache.flink.kubernetes.operator.api.FlinkSessionJob2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinksessionjobs' with unstable version 'v1beta1'2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinkdeployments' with unstable version 'v1beta1'2026-02-06 10:58:30,739 ERROR io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper [] [.] - Informer startup error. Operator will be stopped. Informer: flink.apache.org/v1beta1/flinkdeploymentsjava.util.concurrent.TimeoutException: null at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1892) ~[?:?] at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2027) ~[?:?] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper.start(InformerWrapper.java:88) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerManager.lambda$start$0(InformerManager.java:62) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.api.config.ExecutorServiceManager.lambda$executeAndWaitForAllToComplete$0(ExecutorServiceManager.java:61) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?] at java.lang.Thread.run(Thread.java:829) [?:?]{code}
The log shows the informer startup error occurred 2 minutes after the Operator SDK started. From the [Java Operator SDK doc|https://javadoc.io/static/io.javaoperatorsdk/operator-framework-core/5.0.0/io/javaoperatorsdk/operator/api/config/ConfigurationService.html?utm_source=openai#cacheSyncTimeout()] we can see the default is 2 minutes.
So we want to introduce a new ConfigOption `kubernetes.operator.startup.cache-sync-timeout` to make it configurable.
{code:java}
// KubernetesOperatorConfigOptions
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_CACHE_SYNC_TIMEOUT =
operatorConfig("startup.cache-sync-timeout")
.durationType()
.defaultValue(Duration.ofMinutes(2))
.withDescription("Cache sync timeout for JOSDK informer startup.");{code}
{code:java}
// FlinkOperator#overrideOperatorConfigs()
private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
overrider.withKubernetesClient(client);
var conf = configManager.getDefaultConfig();
var operatorConf = FlinkOperatorConfiguration.fromConfiguration(conf);
int parallelism = operatorConf.getReconcilerMaxParallelism();
if (parallelism == -1) {
LOG.info("Configuring operator with unbounded reconciliation thread pool.");
overrider.withExecutorService(Executors.newCachedThreadPool());
} else {
LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
overrider.withConcurrentReconciliationThreads(parallelism);
}
if (operatorConf.isJosdkMetricsEnabled()) {
overrider.withMetrics(new OperatorJosdkMetrics(metricGroup, configManager));
}
// override cache sync timeout
overrider.withCacheSyncTimeout(conf.get(
KubernetesOperatorConfigOptions.OPERATOR_CACHE_SYNC_TIMEOUT));
// ...
}{code}
I’d like to work on this issue and contribute a PR. If the proposal is accepted, I can open a PR and follow up with the updates.
was:
There are {color:#000000}thousand{color}s Flink deployments in our K8s cluster. When restart the operator, we found the Informer startup error, the log is as follows:
{code:java}
// Operator start log
2026-02-06 10:56:30,688 INFO io.javaoperatorsdk.operator.Operator [] [.] - Operator SDK 4.9.4 (commit: 3588780) built on 2024-08-30T00:52:57.000+0800 starting...2026-02-06 10:56:30,689 INFO io.javaoperatorsdk.operator.Operator [] [.] - Client version: 6.13.22026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkdeploymentcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController, resource: org.apache.flink.kubernetes.operator.api.FlinkDeployment2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkstatesnapshotcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotController, resource: org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinksessionjobcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController, resource: org.apache.flink.kubernetes.operator.api.FlinkSessionJob2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinksessionjobs' with unstable version 'v1beta1'2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinkdeployments' with unstable version 'v1beta1'2026-02-06 10:58:30,739 ERROR io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper [] [.] - Informer startup error. Operator will be stopped. Informer: flink.apache.org/v1beta1/flinkdeploymentsjava.util.concurrent.TimeoutException: null at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1892) ~[?:?] at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2027) ~[?:?] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper.start(InformerWrapper.java:88) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerManager.lambda$start$0(InformerManager.java:62) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.api.config.ExecutorServiceManager.lambda$executeAndWaitForAllToComplete$0(ExecutorServiceManager.java:61) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?] at java.lang.Thread.run(Thread.java:829) [?:?] {code}
The log shows the informer startup error occurred 2 minutes after the Operator SDK started. From the [Java Operator SDK doc|https://javadoc.io/static/io.javaoperatorsdk/operator-framework-core/5.0.0/io/javaoperatorsdk/operator/api/config/ConfigurationService.html?utm_source=openai#cacheSyncTimeout()] we can see the default is 2 minutes.
So we want to introduce a new ConfigOption `kubernetes.operator.startup.cache-sync-timeout` to make it configurable.
{code:java}
// KubernetesOperatorConfigOptions
@Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_CACHE_SYNC_TIMEOUT =
operatorConfig("startup.cache-sync-timeout")
.durationType()
.defaultValue(Duration.ofMinutes(2))
.withDescription("Cache sync timeout for JOSDK informer startup.");{code}
{code:java}
// FlinkOperator#overrideOperatorConfigs()
private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
overrider.withKubernetesClient(client);
var conf = configManager.getDefaultConfig();
var operatorConf = FlinkOperatorConfiguration.fromConfiguration(conf);
int parallelism = operatorConf.getReconcilerMaxParallelism();
if (parallelism == -1) {
LOG.info("Configuring operator with unbounded reconciliation thread pool.");
overrider.withExecutorService(Executors.newCachedThreadPool());
} else {
LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
overrider.withConcurrentReconciliationThreads(parallelism);
}
if (operatorConf.isJosdkMetricsEnabled()) {
overrider.withMetrics(new OperatorJosdkMetrics(metricGroup, configManager));
}
// override cache sync timeout
overrider.withCacheSyncTimeout(conf.get(
KubernetesOperatorConfigOptions.OPERATOR_CACHE_SYNC_TIMEOUT));
// ...
}{code}
I’d like to work on this issue and contribute a PR. If the proposal is accepted, I can open a PR and follow up with the updates.
> Expose JOSDK cacheSyncTimeout as operator config
> ------------------------------------------------
>
> Key: FLINK-39039
> URL: https://issues.apache.org/jira/browse/FLINK-39039
> Project: Flink
> Issue Type: Improvement
> Components: Kubernetes Operator
> Reporter: junzhong qin
> Priority: Not a Priority
>
>
> There are {color:#000000}thousand{color}s Flink deployments in our K8s cluster. When restart the operator, we found the Informer startup error, the log is as follows:
> {code:java}
> // Operator start log
> 2026-02-06 10:56:30,688 INFO io.javaoperatorsdk.operator.Operator [] [.] - Operator SDK 4.9.4 (commit: 3588780) built on 2024-08-30T00:52:57.000+0800 starting...2026-02-06 10:56:30,689 INFO io.javaoperatorsdk.operator.Operator [] [.] - Client version: 6.13.22026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkdeploymentcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController, resource: org.apache.flink.kubernetes.operator.api.FlinkDeployment2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinkstatesnapshotcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotController, resource: org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot2026-02-06 10:56:30,691 INFO io.javaoperatorsdk.operator.processing.Controller [] [.] - Starting 'flinksessionjobcontroller' controller for reconciler: org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController, resource: org.apache.flink.kubernetes.operator.api.FlinkSessionJob2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinksessionjobs' with unstable version 'v1beta1'2026-02-06 10:56:30,734 WARN io.fabric8.kubernetes.client.dsl.internal.VersionUsageUtils [] [.] - The client is using resource type 'flinkdeployments' with unstable version 'v1beta1'2026-02-06 10:58:30,739 ERROR io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper [] [.] - Informer startup error. Operator will be stopped. Informer: flink.apache.org/v1beta1/flinkdeploymentsjava.util.concurrent.TimeoutException: null at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1892) ~[?:?] at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2027) ~[?:?] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerWrapper.start(InformerWrapper.java:88) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.processing.event.source.informer.InformerManager.lambda$start$0(InformerManager.java:62) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at io.javaoperatorsdk.operator.api.config.ExecutorServiceManager.lambda$executeAndWaitForAllToComplete$0(ExecutorServiceManager.java:61) ~[flink-kubernetes-operator-1.11-SNAPSHOT-shaded.jar:1.11-SNAPSHOT] at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?] at java.lang.Thread.run(Thread.java:829) [?:?]{code}
> The log shows the informer startup error occurred 2 minutes after the Operator SDK started. From the [Java Operator SDK doc|https://javadoc.io/static/io.javaoperatorsdk/operator-framework-core/5.0.0/io/javaoperatorsdk/operator/api/config/ConfigurationService.html?utm_source=openai#cacheSyncTimeout()] we can see the default is 2 minutes.
> So we want to introduce a new ConfigOption `kubernetes.operator.startup.cache-sync-timeout` to make it configurable.
>
> {code:java}
> // KubernetesOperatorConfigOptions
> @Documentation.Section(SECTION_ADVANCED)
> public static final ConfigOption<Duration> OPERATOR_CACHE_SYNC_TIMEOUT =
> operatorConfig("startup.cache-sync-timeout")
> .durationType()
> .defaultValue(Duration.ofMinutes(2))
> .withDescription("Cache sync timeout for JOSDK informer startup.");{code}
>
>
>
> {code:java}
> // FlinkOperator#overrideOperatorConfigs()
> private void overrideOperatorConfigs(ConfigurationServiceOverrider overrider) {
> overrider.withKubernetesClient(client);
> var conf = configManager.getDefaultConfig();
> var operatorConf = FlinkOperatorConfiguration.fromConfiguration(conf);
> int parallelism = operatorConf.getReconcilerMaxParallelism();
> if (parallelism == -1) {
> LOG.info("Configuring operator with unbounded reconciliation thread pool.");
> overrider.withExecutorService(Executors.newCachedThreadPool());
> } else {
> LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
> overrider.withConcurrentReconciliationThreads(parallelism);
> }
> if (operatorConf.isJosdkMetricsEnabled()) {
> overrider.withMetrics(new OperatorJosdkMetrics(metricGroup, configManager));
> }
> // override cache sync timeout
> overrider.withCacheSyncTimeout(conf.get(
> KubernetesOperatorConfigOptions.OPERATOR_CACHE_SYNC_TIMEOUT));
> // ...
> }{code}
>
> I’d like to work on this issue and contribute a PR. If the proposal is accepted, I can open a PR and follow up with the updates.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)