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
commit de630f6f8ac0591fab2300fe56d58cdd262002a0 Author: Gyula Fora <[email protected]> AuthorDate: Thu Sep 17 17:24:54 2026 +0200 Revert "[FLINK-35746][Kubernetes-Operator] Add getJobConfiguration and getJobCheckpointConfiguration (#1000)" This reverts commit fdfcc695daba16e4f4a8c9a6c08b7f470787de00. --- .../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, 6 insertions(+), 1108 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 9507973d6..9b9126bc5 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,7 +49,6 @@ 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; @@ -89,17 +88,6 @@ 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("\\.", "\\\\\\.") @@ -144,14 +132,6 @@ 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( @@ -375,56 +355,6 @@ 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 8a0be9081..15c0242d2 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,8 +40,6 @@ 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. */ @@ -58,7 +56,6 @@ 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; @@ -106,65 +103,7 @@ public abstract class FlinkResourceContext<CR extends AbstractFlinkResource<?, ? if (observeConfig != null) { return observeConfig; } - 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; + return observeConfig = createObserveConfig(); } /** 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 423ab1b48..6379e63c1 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,7 +91,6 @@ 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()) { @@ -112,34 +111,6 @@ 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 3be75396f..a32597da8 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 @@ -358,12 +358,7 @@ public abstract class AbstractFlinkResourceReconciler< var scaled = ctx.getFlinkService().scale(ctx, deployConfig); if (scaled) { - var resource = ctx.getResource(); - ctx.getConfigManager() - .invalidateRuntimeConfig( - resource.getMetadata().getNamespace(), - resource.getMetadata().getName()); - ReconciliationUtils.updateStatusForDeployedSpec(resource, deployConfig, clock); + ReconciliationUtils.updateStatusForDeployedSpec(ctx.getResource(), 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 51f3b3be6..e57aa51b9 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,10 +599,6 @@ 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 38e8df1e6..7d08fb015 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 @@ -170,11 +170,6 @@ 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()); @@ -256,12 +251,10 @@ 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(resource, suspendMode, ctx.getObserveConfig()); + var result = + ctx.getFlinkService() + .cancelJob(ctx.getResource(), 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 0c3c1812d..c38afce0c 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,13 +103,6 @@ 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 dd088dd7a..4491ab971 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,15 +76,6 @@ 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, @@ -122,14 +113,10 @@ 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(resource, suspendMode, ctx.getObserveConfig()); + .cancelSessionJob(ctx.getResource(), 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 ad6504236..99753a7e2 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,7 +55,6 @@ 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; @@ -68,14 +67,10 @@ 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; @@ -87,7 +82,6 @@ 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; @@ -929,111 +923,6 @@ 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 64a04e2a2..b1a078fbb 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,12 +133,6 @@ 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 deleted file mode 100644 index eb4642247..000000000 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkRuntimeConfigurationUtils.java +++ /dev/null @@ -1,242 +0,0 @@ -/* - * 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 fac68a616..b17fb624a 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,36 +789,6 @@ 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 d528d8697..715c072b8 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,41 +467,4 @@ 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 f7733010a..61a345df2 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 @@ -1039,354 +1039,6 @@ 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 bb00337c6..abac03963 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,7 +52,6 @@ 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; @@ -61,24 +60,19 @@ 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; @@ -141,10 +135,8 @@ 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; @@ -1573,130 +1565,6 @@ 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;
