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 fdfcc695 [FLINK-35746][Kubernetes-Operator] Add getJobConfiguration
and getJobCheckpointConfiguration (#1000)
fdfcc695 is described below
commit fdfcc695daba16e4f4a8c9a6c08b7f470787de00
Author: Nishita Pattanayak <[email protected]>
AuthorDate: Tue Aug 11 19:40:46 2026 +0530
[FLINK-35746][Kubernetes-Operator] Add getJobConfiguration and
getJobCheckpointConfiguration (#1000)
---
.../operator/config/FlinkConfigManager.java | 70 +++++
.../operator/controller/FlinkResourceContext.java | 63 +++-
.../operator/observer/JobStatusObserver.java | 29 ++
.../AbstractFlinkResourceReconciler.java | 7 +-
.../deployment/AbstractJobReconciler.java | 4 +
.../deployment/ApplicationReconciler.java | 13 +-
.../reconciler/deployment/SessionReconciler.java | 7 +
.../sessionjob/SessionJobReconciler.java | 15 +-
.../operator/service/AbstractFlinkService.java | 111 +++++++
.../kubernetes/operator/service/FlinkService.java | 6 +
.../utils/FlinkRuntimeConfigurationUtils.java | 242 ++++++++++++++
.../kubernetes/operator/TestingFlinkService.java | 30 ++
.../operator/config/FlinkConfigManagerTest.java | 37 +++
.../operator/observer/JobStatusObserverTest.java | 348 +++++++++++++++++++++
.../operator/service/AbstractFlinkServiceTest.java | 132 ++++++++
15 files changed, 1108 insertions(+), 6 deletions(-)
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 9b9126bc..9507973d 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
@@ -49,6 +49,7 @@ import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -88,6 +89,17 @@ public class FlinkConfigManager {
private final Consumer<Set<String>> namespaceListener;
private volatile ConcurrentHashMap<FlinkVersion, List<String>>
relevantFlinkVersionPrefixes;
+ private final Cache<RuntimeConfigCacheKey, Map<String, String>>
runtimeConfigCache;
+
+ /** Cache key for runtime configuration overrides, scoped to a specific
job instance. */
+ @Value
+ @Builder
+ private static class RuntimeConfigCacheKey {
+ String namespace;
+ String name;
+ String jobId;
+ }
+
protected static final Pattern FLINK_VERSION_PATTERN =
Pattern.compile(
VERSION_CONF_PREFIX.replaceAll("\\.", "\\\\\\.")
@@ -132,6 +144,14 @@ public class FlinkConfigManager {
}
});
+ this.runtimeConfigCache =
+ CacheBuilder.newBuilder()
+ .maximumSize(
+ defaultConfig.get(
+
KubernetesOperatorConfigOptions.OPERATOR_CONFIG_CACHE_SIZE))
+ .expireAfterAccess(cacheTimeout)
+ .build();
+
updateDefaultConfig(defaultConfig);
ScheduledExecutorService executorService =
Executors.newSingleThreadScheduledExecutor();
executorService.scheduleWithFixedDelay(
@@ -355,6 +375,56 @@ public class FlinkConfigManager {
return conf;
}
+ private RuntimeConfigCacheKey cacheKey(String namespace, String name,
String jobId) {
+ return
RuntimeConfigCacheKey.builder().namespace(namespace).name(name).jobId(jobId).build();
+ }
+
+ /**
+ * Store runtime configuration overrides in the cache. Called by observers
after fetching
+ * configuration from the Flink REST API.
+ *
+ * @param namespace Resource namespace
+ * @param name Resource name
+ * @param jobId Job ID string
+ * @param config Runtime configuration key-value pairs
+ */
+ public void putRuntimeConfig(
+ String namespace, String name, String jobId, Map<String, String>
config) {
+ runtimeConfigCache.put(
+ cacheKey(namespace, name, jobId),
Collections.unmodifiableMap(config));
+ LOG.debug("Cached runtime configuration with {} entries",
config.size());
+ }
+
+ /**
+ * Get cached runtime configuration overrides for a specific job instance.
+ *
+ * @param namespace Resource namespace
+ * @param name Resource name
+ * @param jobId Job ID string
+ * @return Cached runtime config if present
+ */
+ public Optional<Map<String, String>> getRuntimeConfig(
+ String namespace, String name, String jobId) {
+ return Optional.ofNullable(
+ runtimeConfigCache.getIfPresent(cacheKey(namespace, name,
jobId)));
+ }
+
+ /**
+ * Invalidate all cached runtime configurations for a given resource.
Should be called when a
+ * job is cancelled or an upgrade replaces the running job.
+ *
+ * @param namespace Resource namespace
+ * @param name Resource name
+ */
+ public void invalidateRuntimeConfig(String namespace, String name) {
+ runtimeConfigCache
+ .asMap()
+ .keySet()
+ .removeIf(
+ key -> namespace.equals(key.getNamespace()) &&
name.equals(key.getName()));
+ LOG.debug("Invalidated runtime configuration cache");
+ }
+
private void addOperatorConfigsFromSpec(AbstractFlinkSpec spec,
Configuration conf) {
// Observe config should include the latest operator related settings
if (spec.getFlinkConfiguration() != null) {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
index 15c0242d..8a0be908 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
@@ -40,6 +40,8 @@ import lombok.RequiredArgsConstructor;
import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Optional;
import java.util.function.Function;
/** Context for reconciling a Flink resource. */
@@ -56,6 +58,7 @@ public abstract class FlinkResourceContext<CR extends
AbstractFlinkResource<?, ?
private FlinkOperatorConfiguration operatorConfig;
private Configuration observeConfig;
+ private Optional<Map<String, String>> runtimeConfig;
private FlinkService flinkService;
private KubernetesJobAutoScalerContext autoScalerContext;
@@ -103,7 +106,65 @@ public abstract class FlinkResourceContext<CR extends
AbstractFlinkResource<?, ?
if (observeConfig != null) {
return observeConfig;
}
- return observeConfig = createObserveConfig();
+ observeConfig = createObserveConfig();
+ if (observeConfig != null) {
+ getRuntimeConfig().ifPresent(config ->
config.forEach(observeConfig::setString));
+ }
+ return observeConfig;
+ }
+
+ /**
+ * Get the cached runtime configuration for the current job, memoized per
context to avoid
+ * repeated cache lookups within a single reconciliation cycle. Only a
present result is
+ * memoized so that a later {@link #putRuntimeConfig(Map)} within the same
cycle is visible to
+ * subsequent callers.
+ *
+ * @return Cached runtime config if present.
+ */
+ public Optional<Map<String, String>> getRuntimeConfig() {
+ if (runtimeConfig != null && runtimeConfig.isPresent()) {
+ return runtimeConfig;
+ }
+ var jobStatus = resource.getStatus().getJobStatus();
+ if (jobStatus == null || jobStatus.getJobId() == null) {
+ return Optional.empty();
+ }
+ return runtimeConfig =
+ configManager.getRuntimeConfig(
+ resource.getMetadata().getNamespace(),
+ resource.getMetadata().getName(),
+ jobStatus.getJobId());
+ }
+
+ /**
+ * Store runtime configuration in the global cache and refresh the
context-level memoized view
+ * so subsequent accesses within the same reconciliation cycle avoid an
extra cache lookup.
+ *
+ * @param config Runtime configuration key-value pairs fetched from the
Flink REST API.
+ */
+ public void putRuntimeConfig(Map<String, String> config) {
+ var jobStatus = resource.getStatus().getJobStatus();
+ if (jobStatus == null || jobStatus.getJobId() == null) {
+ return;
+ }
+ configManager.putRuntimeConfig(
+ resource.getMetadata().getNamespace(),
+ resource.getMetadata().getName(),
+ jobStatus.getJobId(),
+ config);
+ runtimeConfig =
+ configManager.getRuntimeConfig(
+ resource.getMetadata().getNamespace(),
+ resource.getMetadata().getName(),
+ jobStatus.getJobId());
+ observeConfig = null;
+ }
+
+ /**
+ * @return The config manager for this context.
+ */
+ public FlinkConfigManager getConfigManager() {
+ return configManager;
}
/**
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
index e205e3b8..b1936481 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -91,6 +91,7 @@ public class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>> {
if (newJobStatusOpt.isPresent()) {
var newJobStatus = newJobStatusOpt.get();
updateJobStatus(ctx, newJobStatus);
+ fetchAndCacheRuntimeConfig(ctx, newJobStatus);
ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
// see if the JM server is up, try to get the exceptions
if (!previousJobStatus.isGloballyTerminalState()) {
@@ -111,6 +112,34 @@ public class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>> {
return false;
}
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ // Skip only globally-terminal states (FINISHED/CANCELED/FAILED) where
the JM REST API is
+ // gone; non-terminal transitional states (INITIALIZING, RESTARTING,
RECONCILING, etc.)
+ // still expose reachable config endpoints and any REST failure is
caught below.
+ if (clusterJobStatus.getJobState().isGloballyTerminalState()) {
+ return;
+ }
+
+ if (ctx.getRuntimeConfig().isPresent()) {
+ LOG.debug("Runtime configuration already cached");
+ return;
+ }
+
+ LOG.debug("Fetching runtime configuration");
+ var jobStatus = ctx.getResource().getStatus().getJobStatus();
+ try {
+ var runtimeConfig =
+ ctx.getFlinkService()
+ .getRuntimeConfiguration(
+ ctx.getObserveConfig(),
+ JobID.fromHexString(jobStatus.getJobId()));
+ ctx.putRuntimeConfig(runtimeConfig);
+ } catch (Exception e) {
+ LOG.warn("Failed to fetch runtime configuration, will retry next
cycle", e);
+ }
+ }
+
/**
* Observe the exceptions raised in the job manager and take appropriate
action.
*
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index 7f8af3ca..3c16b350 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -363,7 +363,12 @@ public abstract class AbstractFlinkResourceReconciler<
var scaled = ctx.getFlinkService().scale(ctx, deployConfig);
if (scaled) {
- ReconciliationUtils.updateStatusForDeployedSpec(ctx.getResource(),
deployConfig, clock);
+ var resource = ctx.getResource();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ resource.getMetadata().getNamespace(),
+ resource.getMetadata().getName());
+ ReconciliationUtils.updateStatusForDeployedSpec(resource,
deployConfig, clock);
}
return scaled;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index e57aa51b..51f3b3be 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -599,6 +599,10 @@ public abstract class AbstractJobReconciler<
protected void resubmitJob(FlinkResourceContext<CR> ctx, boolean
requireHaMetadata)
throws Exception {
LOG.info("Resubmitting Flink job...");
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ ctx.getResource().getMetadata().getNamespace(),
+ ctx.getResource().getMetadata().getName());
SPEC specToRecover =
ReconciliationUtils.getDeployedSpec(ctx.getResource());
var upgradeStatePath =
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 6f0f0e85..99ff5f78 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
@@ -150,6 +150,11 @@ public class ApplicationReconciler
var status = relatedResource.getStatus();
var flinkService = ctx.getFlinkService();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ relatedResource.getMetadata().getNamespace(),
+ relatedResource.getMetadata().getName());
+
ClusterHealthEvaluator.removeLastValidClusterHealthInfo(
relatedResource.getStatus().getClusterInfo());
@@ -230,10 +235,12 @@ public class ApplicationReconciler
@Override
protected boolean cancelJob(FlinkResourceContext<FlinkDeployment> ctx,
SuspendMode suspendMode)
throws Exception {
+ var resource = ctx.getResource();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ resource.getMetadata().getNamespace(),
resource.getMetadata().getName());
var cancelTs = Instant.now();
- var result =
- ctx.getFlinkService()
- .cancelJob(ctx.getResource(), suspendMode,
ctx.getObserveConfig());
+ var result = ctx.getFlinkService().cancelJob(resource, suspendMode,
ctx.getObserveConfig());
result.getSavepointPath()
.ifPresent(location -> setUpgradeSavepointPath(ctx, location,
cancelTs));
return result.isPending();
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 c38afce0..0c3c1812 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
@@ -103,6 +103,13 @@ public class SessionReconciler
boolean requireHaMetadata)
throws Exception {
var cr = ctx.getResource();
+ // Defensive invalidate for consistency with the other deploy()
overrides. In practice this
+ // is a no-op for session-mode FlinkDeployments (jobs live on
FlinkSessionJob resources
+ // with their own cache keys), but it makes the "every deploy()
invalidates" invariant
+ // explicit and safe against future changes to session-cluster caching.
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ cr.getMetadata().getNamespace(),
cr.getMetadata().getName());
setOwnerReference(cr, deployConfig);
ctx.getFlinkService().submitSessionCluster(deployConfig);
cr.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
index 4491ab97..dd088dd7 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
@@ -76,6 +76,15 @@ public class SessionJobReconciler
boolean requireHaMetadata)
throws Exception {
+ // Any deploy path produces a new job submission, so any previously
cached runtime config
+ // for this resource is stale. Defensive invalidate here keeps every
deploy() override
+ // consistent and avoids relying on transitive invalidation by
upstream callers.
+ var relatedResource = ctx.getResource();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ relatedResource.getMetadata().getNamespace(),
+ relatedResource.getMetadata().getName());
+
eventRecorder.triggerEvent(
ctx.getResource(),
EventRecorder.Type.Normal,
@@ -113,10 +122,14 @@ public class SessionJobReconciler
@Override
protected boolean cancelJob(FlinkResourceContext<FlinkSessionJob> ctx,
SuspendMode suspendMode)
throws Exception {
+ var resource = ctx.getResource();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
+ resource.getMetadata().getNamespace(),
resource.getMetadata().getName());
var cancelTs = Instant.now();
var result =
ctx.getFlinkService()
- .cancelSessionJob(ctx.getResource(), suspendMode,
ctx.getObserveConfig());
+ .cancelSessionJob(resource, suspendMode,
ctx.getObserveConfig());
result.getSavepointPath()
.ifPresent(location -> setUpgradeSavepointPath(ctx, location,
cancelTs));
return result.isPending();
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 08467957..bc0e9ccf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -55,6 +55,7 @@ import
org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.ExceptionUtils;
+import
org.apache.flink.kubernetes.operator.utils.FlinkRuntimeConfigurationUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import
org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
@@ -67,10 +68,14 @@ import
org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobConfigHeaders;
+import org.apache.flink.runtime.rest.messages.JobConfigInfo;
import org.apache.flink.runtime.rest.messages.JobExceptionsHeaders;
import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
+import org.apache.flink.runtime.rest.messages.JobMessageParameters;
import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigHeaders;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointIdPathParameter;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointInfo;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatisticDetailsHeaders;
@@ -82,6 +87,7 @@ import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerReque
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
import
org.apache.flink.runtime.rest.messages.job.JobExceptionsMessageParameters;
+import
org.apache.flink.runtime.rest.messages.job.JobManagerJobConfigurationHeaders;
import org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsHeaders;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
@@ -909,6 +915,111 @@ public abstract class AbstractFlinkService implements
FlinkService {
}
}
+ @Override
+ public Map<String, String> getRuntimeConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching runtime configuration");
+ Map<String, String> runtimeConfig = new HashMap<>();
+ boolean fetchFailed = false;
+
+ try {
+ runtimeConfig.putAll(getJobManagerConfiguration(conf, jobId));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.error("Failed to fetch JobManager configuration", e);
+ }
+
+ try {
+ runtimeConfig.putAll(getJobConfiguration(conf, jobId));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.error("Failed to fetch job configuration", e);
+ }
+
+ try {
+ runtimeConfig.putAll(getJobCheckpointConfiguration(conf, jobId));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.error("Failed to fetch checkpoint configuration", e);
+ }
+
+ if (runtimeConfig.isEmpty() && fetchFailed) {
+ throw new RuntimeException("All runtime config REST fetches failed
for job " + jobId);
+ }
+
+ LOG.debug("Fetched merged runtime configuration with {} entries",
runtimeConfig.size());
+ return runtimeConfig;
+ }
+
+ private Map<String, String> getJobManagerConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching JobManager configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ var configurationInfo =
+ clusterClient
+ .sendRequest(
+
JobManagerJobConfigurationHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jmConfig = new HashMap<>();
+ configurationInfo.forEach(entry -> jmConfig.put(entry.getKey(),
entry.getValue()));
+ LOG.debug("Fetched {} JobManager configuration entries",
jmConfig.size());
+ return jmConfig;
+ }
+ }
+
+ private Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders, parameters,
EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ Map<String, String> jobConfig =
+
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configurationInfo);
+ LOG.debug("Fetched {} job configuration entries",
jobConfig.size());
+ return jobConfig;
+ }
+ }
+
+ private Map<String, String> getJobCheckpointConfiguration(Configuration
conf, JobID jobId)
+ throws Exception {
+ LOG.debug("Fetching checkpoint configuration");
+ try (var clusterClient = getClusterClient(conf)) {
+ var checkpointConfigHeaders =
CheckpointConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ var checkpointConfigInfo =
+ clusterClient
+ .sendRequest(
+ checkpointConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+
+ return
FlinkRuntimeConfigurationUtils.mapCheckpointConfiguration(checkpointConfigInfo);
+ }
+ }
+
@VisibleForTesting
protected void runJar(
JobSpec job,
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 b1a078fb..64a04e2a 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
@@ -133,6 +133,12 @@ public interface FlinkService {
AbstractFlinkResource resource, JobID jobId, Configuration
observeConfig)
throws Exception;
+ /**
+ * Fetches the merged runtime configuration for a running job by querying
the JM config, job
+ * execution config, and checkpoint config REST endpoints. Later layers
override earlier ones.
+ */
+ Map<String, String> getRuntimeConfiguration(Configuration conf, JobID
jobId) throws Exception;
+
/** Result of a cancel operation. */
@AllArgsConstructor
class CancelResult {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
new file mode 100644
index 00000000..eb464224
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.utils;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.configuration.StateBackendOptions;
+import org.apache.flink.configuration.StateChangelogOptions;
+import org.apache.flink.runtime.rest.messages.JobConfigInfo;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigInfo;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility class for mapping Flink REST API responses (execution config,
checkpoint config) into
+ * Flink {@link org.apache.flink.configuration.ConfigOption} key-value pairs
suitable for the
+ * runtime configuration cache.
+ */
+public class FlinkRuntimeConfigurationUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FlinkRuntimeConfigurationUtils.class);
+
+ private static final
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind
+ .ObjectMapper
+ CHECKPOINT_CONFIG_MAPPER =
+ new
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind
+ .ObjectMapper();
+
+ /**
+ * Maps REST API JSON field names from {@link CheckpointConfigInfo} to
Flink {@link
+ * org.apache.flink.configuration.ConfigOption} keys. Both sides use
Flink's own constants for
+ * compile-time safety.
+ */
+ public enum CheckpointConfigMapping {
+ PROCESSING_MODE(
+ CheckpointConfigInfo.FIELD_NAME_PROCESSING_MODE,
+ CheckpointingOptions.CHECKPOINTING_CONSISTENCY_MODE.key(),
+ false),
+ INTERVAL(
+ CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_INTERVAL,
+ CheckpointingOptions.CHECKPOINTING_INTERVAL.key(),
+ true),
+ TIMEOUT(
+ CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_TIMEOUT,
+ CheckpointingOptions.CHECKPOINTING_TIMEOUT.key(),
+ true),
+ MIN_PAUSE(
+ CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_MIN_PAUSE,
+ CheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS.key(),
+ true),
+ MAX_CONCURRENT(
+ CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_MAX_CONCURRENT,
+ CheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS.key(),
+ false),
+ TOLERABLE_FAILURES(
+ CheckpointConfigInfo.FIELD_NAME_TOLERABLE_FAILED_CHECKPOINTS,
+ CheckpointingOptions.TOLERABLE_FAILURE_NUMBER.key(),
+ false),
+ UNALIGNED(
+ CheckpointConfigInfo.FIELD_NAME_UNALIGNED_CHECKPOINTS,
+ CheckpointingOptions.ENABLE_UNALIGNED.key(),
+ false),
+ ALIGNED_TIMEOUT(
+ CheckpointConfigInfo.FIELD_NAME_ALIGNED_CHECKPOINT_TIMEOUT,
+ CheckpointingOptions.ALIGNED_CHECKPOINT_TIMEOUT.key(),
+ true),
+ CHECKPOINTS_AFTER_TASKS_FINISH(
+ CheckpointConfigInfo.FIELD_NAME_CHECKPOINTS_AFTER_TASKS_FINISH,
+
CheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH.key(),
+ false),
+ STATE_CHANGELOG(
+ CheckpointConfigInfo.FIELD_NAME_STATE_CHANGELOG,
+ StateChangelogOptions.ENABLE_STATE_CHANGE_LOG.key(),
+ false),
+ PERIODIC_MATERIALIZATION_INTERVAL(
+
CheckpointConfigInfo.FIELD_NAME_PERIODIC_MATERIALIZATION_INTERVAL,
+ StateChangelogOptions.PERIODIC_MATERIALIZATION_INTERVAL.key(),
+ true),
+ CHANGELOG_STORAGE(
+ CheckpointConfigInfo.FIELD_NAME_CHANGELOG_STORAGE,
+ StateChangelogOptions.STATE_CHANGE_LOG_STORAGE.key(),
+ false);
+
+ private final String jsonField;
+ private final String configKey;
+ private final boolean isDuration;
+
+ CheckpointConfigMapping(String jsonField, String configKey, boolean
isDuration) {
+ this.jsonField = jsonField;
+ this.configKey = configKey;
+ this.isDuration = isDuration;
+ }
+
+ public String getJsonField() {
+ return jsonField;
+ }
+ }
+
+ private static final Map<String, String> STATE_BACKEND_NAMES =
+ Map.of(
+ "EmbeddedRocksDBStateBackend", "rocksdb",
+ "HashMapStateBackend", "hashmap");
+
+ private static final Map<String, String> CHECKPOINT_STORAGE_NAMES =
+ Map.of(
+ "FileSystemCheckpointStorage", "filesystem",
+ "JobManagerCheckpointStorage", "jobmanager");
+
+ private FlinkRuntimeConfigurationUtils() {}
+
+ /**
+ * Extract execution configuration (parallelism, object-reuse, global job
parameters) from a
+ * {@link JobConfigInfo} REST response.
+ */
+ public static Map<String, String> mapJobConfiguration(JobConfigInfo
configurationInfo) {
+ Map<String, String> jobConfig = new HashMap<>();
+ if (configurationInfo == null ||
configurationInfo.getExecutionConfigInfo() == null) {
+ return jobConfig;
+ }
+ var execInfo = configurationInfo.getExecutionConfigInfo();
+ jobConfig.put(
+ CoreOptions.DEFAULT_PARALLELISM.key(),
String.valueOf(execInfo.getParallelism()));
+ jobConfig.put(PipelineOptions.OBJECT_REUSE.key(),
String.valueOf(execInfo.isObjectReuse()));
+ jobConfig.putAll(execInfo.getGlobalJobParameters());
+ return jobConfig;
+ }
+
+ /**
+ * Convert a {@link CheckpointConfigInfo} REST response into Flink
configuration key-value
+ * pairs.
+ */
+ public static Map<String, String> mapCheckpointConfiguration(
+ CheckpointConfigInfo checkpointConfigInfo) {
+ Map<String, Object> rawResponse =
+ CHECKPOINT_CONFIG_MAPPER.convertValue(
+ checkpointConfigInfo,
+ new
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.type
+ .TypeReference<
+ Map<String, Object>>() {});
+
+ LOG.debug("Raw checkpoint configuration response: {}", rawResponse);
+ return mapCheckpointFields(rawResponse);
+ }
+
+ private static Map<String, String> mapCheckpointFields(Map<String, Object>
rawResponse) {
+ Map<String, String> mappedConfig = new HashMap<>();
+
+ for (CheckpointConfigMapping mapping :
CheckpointConfigMapping.values()) {
+ if (!rawResponse.containsKey(mapping.jsonField)) {
+ continue;
+ }
+ String value = String.valueOf(rawResponse.get(mapping.jsonField));
+
+ if (mapping == CheckpointConfigMapping.PROCESSING_MODE) {
+ value = value.toUpperCase();
+ }
+ if (mapping.isDuration) {
+ value += "ms";
+ }
+
+ mappedConfig.put(mapping.configKey, value);
+ }
+
+ mapExternalizedCheckpointInfo(rawResponse, mappedConfig);
+ mapStateBackendAndStorage(rawResponse, mappedConfig);
+
+ LOG.debug(
+ "Mapped {} checkpoint configuration entries: {}",
+ mappedConfig.size(),
+ mappedConfig);
+ return mappedConfig;
+ }
+
+ private static void mapExternalizedCheckpointInfo(
+ Map<String, Object> rawResponse, Map<String, String> mappedConfig)
{
+ Object externalizationObj =
+
rawResponse.get(CheckpointConfigInfo.FIELD_NAME_EXTERNALIZED_CHECKPOINT_CONFIG);
+ if (externalizationObj instanceof Map) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> externalization = (Map<String, Object>)
externalizationObj;
+ Boolean enabled =
+ (Boolean)
+ externalization.get(
+
CheckpointConfigInfo.ExternalizedCheckpointInfo
+ .FIELD_NAME_ENABLED);
+ Boolean deleteOnCancellation =
+ (Boolean)
+ externalization.get(
+
CheckpointConfigInfo.ExternalizedCheckpointInfo
+
.FIELD_NAME_DELETE_ON_CANCELLATION);
+
+ String retention = "NO_EXTERNALIZED_CHECKPOINTS";
+ if (Boolean.TRUE.equals(enabled)) {
+ retention =
+ Boolean.TRUE.equals(deleteOnCancellation)
+ ? "DELETE_ON_CANCELLATION"
+ : "RETAIN_ON_CANCELLATION";
+ }
+ mappedConfig.put(
+
CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION.key(), retention);
+ }
+ }
+
+ private static void mapStateBackendAndStorage(
+ Map<String, Object> rawResponse, Map<String, String> mappedConfig)
{
+ if
(rawResponse.containsKey(CheckpointConfigInfo.FIELD_NAME_STATE_BACKEND)) {
+ String raw =
+
String.valueOf(rawResponse.get(CheckpointConfigInfo.FIELD_NAME_STATE_BACKEND));
+ mappedConfig.put(
+ StateBackendOptions.STATE_BACKEND.key(),
+ STATE_BACKEND_NAMES.getOrDefault(raw, raw.toLowerCase()));
+ }
+ if
(rawResponse.containsKey(CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_STORAGE)) {
+ String raw =
+ String.valueOf(
+
rawResponse.get(CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_STORAGE));
+ mappedConfig.put(
+ CheckpointingOptions.CHECKPOINT_STORAGE.key(),
+ CHECKPOINT_STORAGE_NAMES.getOrDefault(raw,
raw.toLowerCase()));
+ }
+ }
+}
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 b17fb624..fac68a61 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
@@ -789,6 +789,36 @@ public class TestingFlinkService extends
AbstractFlinkService {
jobExceptionsMap.put(jobId, newExceptionHistory);
}
+ private final Map<JobID, Map<String, String>> runtimeJmConfigs = new
HashMap<>();
+ private final Map<JobID, Map<String, String>> runtimeJobConfigs = new
HashMap<>();
+ private final Map<JobID, Map<String, String>> runtimeCheckpointConfigs =
new HashMap<>();
+ @Setter private Exception runtimeConfigFetchException;
+
+ public void setRuntimeJmConfig(JobID jobId, Map<String, String> config) {
+ runtimeJmConfigs.put(jobId, new HashMap<>(config));
+ }
+
+ public void setRuntimeJobConfig(JobID jobId, Map<String, String> config) {
+ runtimeJobConfigs.put(jobId, new HashMap<>(config));
+ }
+
+ public void setRuntimeCheckpointConfig(JobID jobId, Map<String, String>
config) {
+ runtimeCheckpointConfigs.put(jobId, new HashMap<>(config));
+ }
+
+ @Override
+ public Map<String, String> getRuntimeConfiguration(Configuration conf,
JobID jobId)
+ throws Exception {
+ if (runtimeConfigFetchException != null) {
+ throw runtimeConfigFetchException;
+ }
+ Map<String, String> merged = new HashMap<>();
+ merged.putAll(runtimeJmConfigs.getOrDefault(jobId, new HashMap<>()));
+ merged.putAll(runtimeJobConfigs.getOrDefault(jobId, new HashMap<>()));
+ merged.putAll(runtimeCheckpointConfigs.getOrDefault(jobId, new
HashMap<>()));
+ return merged;
+ }
+
public void setSavepointTriggerException(Exception exception) {
this.savepointTriggerException = exception;
}
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 715c072b..d528d869 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
@@ -467,4 +467,41 @@ public class FlinkConfigManagerTest {
assertTrue(completed2.get());
assertTrue(completed3.get());
}
+
+ @Test
+ public void testRuntimeConfigCachePutAndGet() {
+ var configManager = new FlinkConfigManager(new Configuration());
+
+ assertFalse(configManager.getRuntimeConfig("ns", "app",
"job1").isPresent());
+
+ Map<String, String> runtimeConf = Map.of("parallelism.default", "4",
"key1", "val1");
+ configManager.putRuntimeConfig("ns", "app", "job1", runtimeConf);
+
+ var cached = configManager.getRuntimeConfig("ns", "app", "job1");
+ assertTrue(cached.isPresent());
+ assertEquals("4", cached.get().get("parallelism.default"));
+ assertEquals("val1", cached.get().get("key1"));
+
+ assertFalse(configManager.getRuntimeConfig("ns", "app",
"job2").isPresent());
+ assertFalse(configManager.getRuntimeConfig("ns", "other",
"job1").isPresent());
+ }
+
+ @Test
+ public void testRuntimeConfigInvalidate() {
+ var configManager = new FlinkConfigManager(new Configuration());
+
+ configManager.putRuntimeConfig("ns", "app", "job1", Map.of("k1",
"v1"));
+ configManager.putRuntimeConfig("ns", "app", "job2", Map.of("k2",
"v2"));
+ configManager.putRuntimeConfig("ns", "other", "job3", Map.of("k3",
"v3"));
+
+ assertTrue(configManager.getRuntimeConfig("ns", "app",
"job1").isPresent());
+ assertTrue(configManager.getRuntimeConfig("ns", "app",
"job2").isPresent());
+ assertTrue(configManager.getRuntimeConfig("ns", "other",
"job3").isPresent());
+
+ configManager.invalidateRuntimeConfig("ns", "app");
+
+ assertFalse(configManager.getRuntimeConfig("ns", "app",
"job1").isPresent());
+ assertFalse(configManager.getRuntimeConfig("ns", "app",
"job2").isPresent());
+ assertTrue(configManager.getRuntimeConfig("ns", "other",
"job3").isPresent());
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java
index 67a0a929..e0cca019 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserverTest.java
@@ -956,6 +956,354 @@ public class JobStatusObserverTest extends
OperatorTestBase {
return job;
}
+ @Test
+ public void testRuntimeConfigFetchAndCache() throws Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ // Configure the mock service to return runtime config
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+ flinkService.setRuntimeJobConfig(
+ jobId,
+ Map.of("parallelism.default", "8",
"execution.checkpointing.interval", "60000"));
+ flinkService.setRuntimeCheckpointConfig(
+ jobId, Map.of("execution.checkpointing.mode", "EXACTLY_ONCE"));
+
+ // Verify runtime config is not cached before observation
+ assertFalse(
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId())
+ .isPresent());
+
+ // Observe the job - this should fetch and cache runtime config
+ observer.observe(ctx);
+
+ // Verify runtime config is now cached
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(cachedConfig.isPresent());
+ assertEquals("8", cachedConfig.get().get("parallelism.default"));
+ assertEquals("60000",
cachedConfig.get().get("execution.checkpointing.interval"));
+ assertEquals("EXACTLY_ONCE",
cachedConfig.get().get("execution.checkpointing.mode"));
+ }
+
+ @Test
+ public void testRuntimeConfigNotFetchedWhenJobNotOnCluster() throws
Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.CREATED);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ // Don't submit the job to the cluster -- observe() won't find a
matching job,
+ // so fetchAndCacheRuntimeConfig should never be called.
+ observer.observe(ctx);
+
+ assertFalse(
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId())
+ .isPresent());
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = JobStatus.class,
+ names = {"FINISHED", "CANCELED", "FAILED"})
+ public void testRuntimeConfigNotFetchedWhenGloballyTerminal(JobStatus
terminalState)
+ throws Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(terminalState);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+ var job =
+ flinkService.listJobs().stream()
+ .filter(t -> t.f1.getJobId().equals(jobId))
+ .findAny()
+ .orElseThrow();
+ job.f1 = new JobStatusMessage(jobId, job.f1.getJobName(),
terminalState, 0);
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"8"));
+
+ observer.observe(ctx);
+
+ assertFalse(
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId())
+ .isPresent());
+ }
+
+ @Test
+ public void testRuntimeConfigMemoizedWithinReconciliationCycle() throws
Exception {
+ var deployment = initDeployment();
+ var jobStatus = deployment.getStatus().getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"4"));
+
+ // Populate the memoized context-level view.
+ observer.observe(ctx);
+
+ var firstView = ctx.getRuntimeConfig();
+ assertTrue(firstView.isPresent());
+
+ // Mutate the global cache underneath the context. If the context is
honouring its memo,
+ // subsequent getRuntimeConfig() calls must NOT observe this change
within the same cycle.
+ ctx.getConfigManager()
+ .putRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId(),
+ Map.of("parallelism.default", "999"));
+
+ assertEquals("4",
ctx.getRuntimeConfig().get().get("parallelism.default"));
+ }
+
+ @Test
+ public void testRuntimeConfigNotFetchedWhenAlreadyCached() throws
Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+
+ // Pre-populate the cache
+ ctx.getConfigManager()
+ .putRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId(),
+ Map.of("pre-cached", "value"));
+
+ // Configure mock to return different values
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"999"));
+
+ // Observe the job
+ observer.observe(ctx);
+
+ // Verify the cached config is unchanged (not refetched)
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(cachedConfig.isPresent());
+ assertEquals("value", cachedConfig.get().get("pre-cached"));
+ assertFalse(cachedConfig.get().containsKey("parallelism.default"));
+ }
+
+ @Test
+ public void testObserveConfigIncludesRuntimeOverrides() throws Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ // Set user configuration
+
deployment.getSpec().getFlinkConfiguration().put("parallelism.default", "2");
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+
+ // Configure runtime config that overrides user config
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"10"));
+
+ // Observe the job - this caches runtime config
+ observer.observe(ctx);
+
+ // Get observe config - should include runtime overrides
+ var observeConfig = ctx.getObserveConfig();
+ assertEquals("10", observeConfig.getString("parallelism.default",
null));
+ }
+
+ @Test
+ public void testSessionJobRuntimeConfigFetchAndCache() throws Exception {
+ var sessionJob = initSessionJob();
+ var status = sessionJob.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
+ getResourceContext(
+ sessionJob,
+
TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitJobToSessionCluster(
+ sessionJob.getMetadata(),
+ sessionJob.getSpec(),
+ jobId,
+ ctx.getDeployConfig(sessionJob.getSpec()),
+ null);
+
+ // Configure the mock service to return runtime config
+ flinkService.setRuntimeJobConfig(
+ jobId, Map.of("parallelism.default", "4", "restart-strategy",
"fixed-delay"));
+ flinkService.setRuntimeCheckpointConfig(
+ jobId, Map.of("execution.checkpointing.interval", "30000"));
+
+ // Observe the job - this should fetch and cache runtime config
+ observer.observe(ctx);
+
+ // Verify runtime config is cached
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ sessionJob.getMetadata().getNamespace(),
+ sessionJob.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(cachedConfig.isPresent());
+ assertEquals("4", cachedConfig.get().get("parallelism.default"));
+ assertEquals("fixed-delay",
cachedConfig.get().get("restart-strategy"));
+ assertEquals("30000",
cachedConfig.get().get("execution.checkpointing.interval"));
+ }
+
+ @Test
+ public void testRuntimeConfigFetchFailureDoesNotBreakObservation() throws
Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+
+ // Configure mock to throw exception when fetching runtime config
+ flinkService.setRuntimeConfigFetchException(new Exception("REST API
unavailable"));
+
+ // Observe should not fail even if runtime config fetch fails
+ boolean result = observer.observe(ctx);
+ assertTrue(result);
+
+ // When both fetches fail, nothing should be cached so we retry next
cycle
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertFalse(cachedConfig.isPresent());
+
+ // Clear the exception and observe again -- should now fetch and cache
successfully
+ flinkService.setRuntimeConfigFetchException(null);
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"4"));
+ flinkService.setRuntimeCheckpointConfig(
+ jobId, Map.of("execution.checkpointing.interval", "5000"));
+
+ observer.observe(ctx);
+
+ var retriedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(retriedConfig.isPresent());
+ assertEquals("4", retriedConfig.get().get("parallelism.default"));
+ assertEquals("5000",
retriedConfig.get().get("execution.checkpointing.interval"));
+ }
+
+ @Test
+ public void testPartialFetchFailureCachesSuccessfulPortion() throws
Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+
+ // Only set job config (no checkpoint config) -- checkpoint fetch will
return empty map
+ // but won't throw, so this tests the case where one succeeds with data
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"6"));
+
+ observer.observe(ctx);
+
+ // Should be cached since at least one fetch returned data
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(cachedConfig.isPresent());
+ assertEquals("6", cachedConfig.get().get("parallelism.default"));
+ }
+
+ @Test
+ public void testCachedRuntimeConfigIsImmutable() throws Exception {
+ var deployment = initDeployment();
+ var status = deployment.getStatus();
+ var jobStatus = status.getJobStatus();
+ jobStatus.setState(JobStatus.RUNNING);
+
+ FlinkResourceContext<AbstractFlinkResource<?, ?>> ctx =
getResourceContext(deployment);
+
+ var jobId = JobID.fromHexString(jobStatus.getJobId());
+ flinkService.submitApplicationCluster(
+ deployment.getSpec().getJob(),
ctx.getDeployConfig(deployment.getSpec()), false);
+ flinkService.setRuntimeJobConfig(jobId, Map.of("parallelism.default",
"4"));
+
+ observer.observe(ctx);
+
+ var cachedConfig =
+ ctx.getConfigManager()
+ .getRuntimeConfig(
+ deployment.getMetadata().getNamespace(),
+ deployment.getMetadata().getName(),
+ jobStatus.getJobId());
+ assertTrue(cachedConfig.isPresent());
+
+ // Attempting to mutate the cached map should throw
+ var configMap = cachedConfig.get();
+ org.junit.jupiter.api.Assertions.assertThrows(
+ UnsupportedOperationException.class, () ->
configMap.put("new.key", "value"));
+ }
+
private void simulateAsyncSuspendInProgress(AbstractFlinkResource<?, ?>
resource) {
// A stateless or last-state session job suspend goes through an
asynchronous cancellation,
// the reconciler only initiates it, records UPGRADING and waits for
the observer
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 064c0926..a6f3ed69 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -52,6 +52,7 @@ import
org.apache.flink.kubernetes.operator.exception.UpgradeFailureException;
import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import
org.apache.flink.kubernetes.operator.utils.FlinkRuntimeConfigurationUtils;
import org.apache.flink.runtime.checkpoint.CheckpointStatsStatus;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -60,19 +61,24 @@ import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
+import org.apache.flink.runtime.rest.messages.ConfigurationInfo;
+import org.apache.flink.runtime.rest.messages.ConfigurationInfoEntry;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.JobConfigInfo;
import org.apache.flink.runtime.rest.messages.JobExceptionsInfoWithHistory;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.MessageParameters;
import org.apache.flink.runtime.rest.messages.RequestBody;
import org.apache.flink.runtime.rest.messages.ResponseBody;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigInfo;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointInfo;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatistics;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatusMessageParameters;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointTriggerMessageParameters;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
import
org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
+import
org.apache.flink.runtime.rest.messages.job.JobManagerJobConfigurationHeaders;
import
org.apache.flink.runtime.rest.messages.job.metrics.JobMetricsMessageParameters;
import org.apache.flink.runtime.rest.messages.job.metrics.Metric;
import
org.apache.flink.runtime.rest.messages.job.metrics.MetricCollectionResponseBody;
@@ -135,8 +141,10 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
@@ -1537,6 +1545,130 @@ public class AbstractFlinkServiceTest {
.readValue(jsonParser, JobExceptionsInfoWithHistory.class);
}
+ @Test
+ void testAllCheckpointConfigInfoFieldNamesCoveredByMapping() throws
Exception {
+ Set<String> declared =
discoverFieldNameConstants(CheckpointConfigInfo.class);
+ assertFalse(declared.isEmpty(), "Should discover FIELD_NAME_*
constants");
+
+ Set<String> handled = new HashSet<>();
+ for (FlinkRuntimeConfigurationUtils.CheckpointConfigMapping m :
+
FlinkRuntimeConfigurationUtils.CheckpointConfigMapping.values()) {
+ handled.add(m.getJsonField());
+ }
+
handled.add(CheckpointConfigInfo.FIELD_NAME_EXTERNALIZED_CHECKPOINT_CONFIG);
+ handled.add(CheckpointConfigInfo.FIELD_NAME_STATE_BACKEND);
+ handled.add(CheckpointConfigInfo.FIELD_NAME_CHECKPOINT_STORAGE);
+
+ declared.removeAll(handled);
+ assertTrue(
+ declared.isEmpty(),
+ "CheckpointConfigInfo FIELD_NAME constants not covered by "
+ + "CheckpointConfigMapping or explicit handlers: "
+ + declared);
+ }
+
+ @Test
+ void testAllExecutionConfigInfoFieldNamesCoveredByMapping() throws
Exception {
+ Set<String> declared =
discoverFieldNameConstants(JobConfigInfo.ExecutionConfigInfo.class);
+ assertFalse(declared.isEmpty(), "Should discover FIELD_NAME_*
constants");
+
+ Set<String> handled =
+ new HashSet<>(
+ Set.of(
+
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_PARALLELISM,
+
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_OBJECT_REUSE_MODE,
+
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_GLOBAL_JOB_PARAMETERS,
+ // Informational only, not runtime config
overrides
+
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_EXECUTION_MODE,
+
JobConfigInfo.ExecutionConfigInfo.FIELD_NAME_RESTART_STRATEGY));
+
+ declared.removeAll(handled);
+ assertTrue(
+ declared.isEmpty(),
+ "ExecutionConfigInfo FIELD_NAME constants not covered: "
+ + declared
+ + ". Add mapping or exclude explicitly.");
+ }
+
+ @Test
+ void testMapJobConfigurationMapsAllExpectedFields() {
+ JobConfigInfo configInfo =
+ new JobConfigInfo(
+ new JobID(),
+ "test-job",
+ new JobConfigInfo.ExecutionConfigInfo(
+ "PIPELINED",
+ "fixedDelay",
+ 4,
+ true,
+ Map.of("user.param", "value1")));
+
+ Map<String, String> result =
FlinkRuntimeConfigurationUtils.mapJobConfiguration(configInfo);
+
+ assertEquals("4", result.get("parallelism.default"));
+ assertEquals("true", result.get("pipeline.object-reuse"));
+ assertEquals("value1", result.get("user.param"));
+ assertEquals(3, result.size());
+ }
+
+ @Test
+ void testMapJobConfigurationHandlesNullGracefully() {
+
assertTrue(FlinkRuntimeConfigurationUtils.mapJobConfiguration(null).isEmpty());
+ assertTrue(
+ FlinkRuntimeConfigurationUtils.mapJobConfiguration(
+ new JobConfigInfo(new JobID(), "test-job",
null))
+ .isEmpty());
+ }
+
+ @Test
+ void testGetRuntimeConfigurationIncludesJmConfig() throws Exception {
+ var jmConfig = new ConfigurationInfo();
+ jmConfig.add(new ConfigurationInfoEntry("jobmanager.scheduler",
"Adaptive"));
+ jmConfig.add(new ConfigurationInfoEntry("state.savepoints.dir",
"/s3/savepoints"));
+ jmConfig.add(new ConfigurationInfoEntry("high-availability",
"ZOOKEEPER"));
+
+ var service =
+ getTestingService(
+ (headers, params, body) -> {
+ if (headers instanceof
JobManagerJobConfigurationHeaders) {
+ return
CompletableFuture.completedFuture((ResponseBody) jmConfig);
+ }
+ return CompletableFuture.failedFuture(
+ new UnsupportedOperationException());
+ });
+
+ var result = service.getRuntimeConfiguration(configuration,
JobID.generate());
+ assertEquals("Adaptive", result.get("jobmanager.scheduler"));
+ assertEquals("/s3/savepoints", result.get("state.savepoints.dir"));
+ assertEquals("ZOOKEEPER", result.get("high-availability"));
+ }
+
+ @Test
+ void testGetRuntimeConfigurationThrowsWhenAllFetchesFail() throws
Exception {
+ var service =
+ getTestingService(
+ (headers, params, body) ->
+ CompletableFuture.failedFuture(
+ new RuntimeException("connection
refused")));
+
+ assertThrows(
+ RuntimeException.class,
+ () -> service.getRuntimeConfiguration(configuration,
JobID.generate()));
+ }
+
+ /** Discovers all {@code FIELD_NAME_*} string constants on a class via
reflection. */
+ private static Set<String> discoverFieldNameConstants(Class<?> clazz)
throws Exception {
+ Set<String> names = new HashSet<>();
+ for (java.lang.reflect.Field f : clazz.getDeclaredFields()) {
+ if (f.getName().startsWith("FIELD_NAME_")
+ && f.getType() == String.class
+ && java.lang.reflect.Modifier.isStatic(f.getModifiers())) {
+ names.add((String) f.get(null));
+ }
+ }
+ return names;
+ }
+
class TestingService extends AbstractFlinkService {
RestClusterClient<String> clusterClient;
