gyfora commented on code in PR #1000:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1000#discussion_r3056247456
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -140,6 +161,12 @@ public Configuration load(Key k) {
cacheTimeout.toMillis(),
TimeUnit.MILLISECONDS);
+ executorService.scheduleWithFixedDelay(
+ runtimeConfigCache::cleanUp,
+ cacheTimeout.toMillis(),
+ cacheTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
Review Comment:
Is this necessary? I think we probably don't need it for the other cache
either as we are continuously using the cache itself
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -351,6 +378,89 @@ public Configuration getObserveConfig(FlinkDeployment
deployment) {
return conf;
}
+ /**
+ * 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) {
+ RuntimeConfigCacheKey cacheKey =
+ RuntimeConfigCacheKey.builder()
+ .namespace(namespace)
+ .name(name)
+ .jobId(jobId)
+ .build();
+ runtimeConfigCache.put(cacheKey, Collections.unmodifiableMap(config));
+ LOG.debug(
+ "Cached runtime configuration for job {} (resource {}/{})",
jobId, namespace, name);
Review Comment:
no need to put namespace/name in the log as it's already part of the MDC
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,85 @@ public boolean observe(FlinkResourceContext<R> ctx) {
return false;
}
+ /**
+ * Fetch runtime configuration from the Flink REST API and cache it in the
config manager when
+ * the job is in RUNNING state and no cached config exists for the current
job ID. After
+ * caching, the memoized observe config is cleared so subsequent consumers
see the merged
+ * values.
+ */
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ if (clusterJobStatus.getJobState() != JobStatus.RUNNING) {
+ return;
+ }
+
+ var resource = ctx.getResource();
+ var jobStatus = resource.getStatus().getJobStatus();
+ String namespace = resource.getMetadata().getNamespace();
+ String name = resource.getMetadata().getName();
+ String jobId = jobStatus.getJobId();
+
+ if (jobId == null) {
+ return;
+ }
+
+ var configManager = ctx.getConfigManager();
+
+ if (configManager.getRuntimeConfig(namespace, name,
jobId).isPresent()) {
+ LOG.debug("Runtime configuration already cached for job {}",
jobId);
+ return;
+ }
+
+ LOG.debug("Fetching runtime configuration for job {}", jobId);
+ var conf = ctx.getObserveConfig();
+ var flinkService = ctx.getFlinkService();
+ var jobIdObj = JobID.fromHexString(jobId);
+ Map<String, String> runtimeConfig = new HashMap<>();
+ boolean fetchFailed = false;
+
+ try {
+ Map<String, String> jobConfig =
flinkService.getJobConfiguration(conf, jobIdObj);
+ if (jobConfig != null) {
+ runtimeConfig.putAll(jobConfig);
+ LOG.debug(
+ "Fetched {} job configuration entries for job {}",
jobConfig.size(), jobId);
+ }
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch job configuration for job {}: {}",
jobId, e.getMessage());
Review Comment:
can we pass the exception to the warn ?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentContext.java:
##########
@@ -59,7 +59,9 @@ public Configuration getDeployConfig(AbstractFlinkSpec spec) {
@Override
protected Configuration createObserveConfig() {
- return configManager.getObserveConfig(getResource());
+ Configuration conf = configManager.getObserveConfig(getResource());
Review Comment:
should getObserveConfig automatically apply the overrides in the
configmanager?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -351,6 +378,89 @@ public Configuration getObserveConfig(FlinkDeployment
deployment) {
return conf;
}
+ /**
+ * 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) {
+ RuntimeConfigCacheKey cacheKey =
+ RuntimeConfigCacheKey.builder()
+ .namespace(namespace)
+ .name(name)
+ .jobId(jobId)
+ .build();
+ runtimeConfigCache.put(cacheKey, Collections.unmodifiableMap(config));
+ LOG.debug(
+ "Cached runtime configuration for job {} (resource {}/{})",
jobId, namespace, name);
+ }
+
+ /**
+ * 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) {
+ RuntimeConfigCacheKey cacheKey =
+ RuntimeConfigCacheKey.builder()
+ .namespace(namespace)
+ .name(name)
+ .jobId(jobId)
+ .build();
+ return Optional.ofNullable(runtimeConfigCache.getIfPresent(cacheKey));
+ }
+
+ /**
+ * 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 for resource
{}/{}", namespace, name);
Review Comment:
no need to put namespace/name in the log as it's already part of the MDC
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,85 @@ public boolean observe(FlinkResourceContext<R> ctx) {
return false;
}
+ /**
+ * Fetch runtime configuration from the Flink REST API and cache it in the
config manager when
+ * the job is in RUNNING state and no cached config exists for the current
job ID. After
+ * caching, the memoized observe config is cleared so subsequent consumers
see the merged
+ * values.
+ */
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ if (clusterJobStatus.getJobState() != JobStatus.RUNNING) {
+ return;
+ }
+
+ var resource = ctx.getResource();
+ var jobStatus = resource.getStatus().getJobStatus();
+ String namespace = resource.getMetadata().getNamespace();
+ String name = resource.getMetadata().getName();
+ String jobId = jobStatus.getJobId();
+
+ if (jobId == null) {
+ return;
+ }
+
+ var configManager = ctx.getConfigManager();
+
+ if (configManager.getRuntimeConfig(namespace, name,
jobId).isPresent()) {
+ LOG.debug("Runtime configuration already cached for job {}",
jobId);
+ return;
+ }
+
+ LOG.debug("Fetching runtime configuration for job {}", jobId);
+ var conf = ctx.getObserveConfig();
+ var flinkService = ctx.getFlinkService();
+ var jobIdObj = JobID.fromHexString(jobId);
+ Map<String, String> runtimeConfig = new HashMap<>();
+ boolean fetchFailed = false;
+
+ try {
+ Map<String, String> jobConfig =
flinkService.getJobConfiguration(conf, jobIdObj);
+ if (jobConfig != null) {
+ runtimeConfig.putAll(jobConfig);
+ LOG.debug(
+ "Fetched {} job configuration entries for job {}",
jobConfig.size(), jobId);
+ }
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch job configuration for job {}: {}",
jobId, e.getMessage());
+ LOG.debug("Job configuration fetch exception details", e);
+ }
+
+ try {
+ Map<String, String> checkpointConfig =
+ flinkService.getJobCheckpointConfiguration(conf, jobIdObj);
+ if (checkpointConfig != null) {
+ runtimeConfig.putAll(checkpointConfig);
+ LOG.debug(
+ "Fetched {} checkpoint configuration entries for job
{}",
+ checkpointConfig.size(),
+ jobId);
+ }
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn(
+ "Failed to fetch checkpoint configuration for job {}: {}",
+ jobId,
+ e.getMessage());
+ LOG.debug("Checkpoint configuration fetch exception details", e);
+ }
+
+ if (runtimeConfig.isEmpty() && fetchFailed) {
+ LOG.warn(
+ "All runtime config REST fetches failed for job {}, will
retry next cycle",
Review Comment:
I think in general we can remove all job ids also from the logs, I don't
think that adds any info
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +884,308 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration for job {}", jobId);
+ try (var clusterClient = getClusterClient(conf)) {
+ Map<String, String> jobConfig = new HashMap<>();
+
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+ if (configurationInfo != null) {
+ if (configurationInfo.getExecutionConfigInfo() != null) {
+ jobConfig.put(
+ "parallelism.default",
Review Comment:
We should use the config options
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,85 @@ public boolean observe(FlinkResourceContext<R> ctx) {
return false;
}
+ /**
+ * Fetch runtime configuration from the Flink REST API and cache it in the
config manager when
+ * the job is in RUNNING state and no cached config exists for the current
job ID. After
+ * caching, the memoized observe config is cleared so subsequent consumers
see the merged
+ * values.
+ */
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ if (clusterJobStatus.getJobState() != JobStatus.RUNNING) {
+ return;
+ }
+
+ var resource = ctx.getResource();
+ var jobStatus = resource.getStatus().getJobStatus();
+ String namespace = resource.getMetadata().getNamespace();
+ String name = resource.getMetadata().getName();
+ String jobId = jobStatus.getJobId();
+
+ if (jobId == null) {
+ return;
+ }
+
+ var configManager = ctx.getConfigManager();
+
+ if (configManager.getRuntimeConfig(namespace, name,
jobId).isPresent()) {
+ LOG.debug("Runtime configuration already cached for job {}",
jobId);
+ return;
+ }
+
+ LOG.debug("Fetching runtime configuration for job {}", jobId);
+ var conf = ctx.getObserveConfig();
+ var flinkService = ctx.getFlinkService();
+ var jobIdObj = JobID.fromHexString(jobId);
+ Map<String, String> runtimeConfig = new HashMap<>();
+ boolean fetchFailed = false;
+
+ try {
+ Map<String, String> jobConfig =
flinkService.getJobConfiguration(conf, jobIdObj);
+ if (jobConfig != null) {
+ runtimeConfig.putAll(jobConfig);
+ LOG.debug(
+ "Fetched {} job configuration entries for job {}",
jobConfig.size(), jobId);
+ }
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch job configuration for job {}: {}",
jobId, e.getMessage());
+ LOG.debug("Job configuration fetch exception details", e);
+ }
+
+ try {
+ Map<String, String> checkpointConfig =
+ flinkService.getJobCheckpointConfiguration(conf, jobIdObj);
+ if (checkpointConfig != null) {
+ runtimeConfig.putAll(checkpointConfig);
+ LOG.debug(
+ "Fetched {} checkpoint configuration entries for job
{}",
+ checkpointConfig.size(),
+ jobId);
+ }
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn(
+ "Failed to fetch checkpoint configuration for job {}: {}",
+ jobId,
+ e.getMessage());
Review Comment:
pass exception instead of message
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +884,308 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration for job {}", jobId);
+ try (var clusterClient = getClusterClient(conf)) {
+ Map<String, String> jobConfig = new HashMap<>();
+
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+ if (configurationInfo != null) {
+ if (configurationInfo.getExecutionConfigInfo() != null) {
+ jobConfig.put(
+ "parallelism.default",
+ String.valueOf(
+ configurationInfo
+ .getExecutionConfigInfo()
+ .getParallelism()));
+ jobConfig.put(
+ "pipeline.object-reuse",
+ String.valueOf(
+ configurationInfo
+ .getExecutionConfigInfo()
+ .isObjectReuse()));
+ jobConfig.putAll(
+ configurationInfo
+ .getExecutionConfigInfo()
+ .getGlobalJobParameters());
+ }
+
+ LOG.debug("Fetched {} job configuration entries",
jobConfig.size());
+ } else {
+ LOG.warn("Job configuration is null for job {}", jobId);
+ }
+
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to fetch job configuration for job {} via REST
API: {}",
+ jobId,
+ e.getMessage());
+ LOG.debug("Job configuration fetch exception details", e);
+ }
+
+ LOG.debug(
+ "Fetched {} configuration entries for job {}: {}",
+ jobConfig.size(),
+ jobId,
+ jobConfig);
+ return jobConfig;
+ }
+ }
+
+ /**
+ * 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.
+ */
+ private enum CheckpointConfigMapping {
Review Comment:
We should move all the mapping logic for the Execution + Checkpoint config
to a new utility class.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +884,308 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @Override
+ public Map<String, String> getJobConfiguration(Configuration conf, JobID
jobId)
+ throws Exception {
+ LOG.debug("Fetching job configuration for job {}", jobId);
+ try (var clusterClient = getClusterClient(conf)) {
+ Map<String, String> jobConfig = new HashMap<>();
+
+ var jobConfigHeaders = JobConfigHeaders.getInstance();
+ var parameters = new JobMessageParameters();
+ parameters.jobPathParameter.resolve(jobId);
+
+ try {
+ JobConfigInfo configurationInfo =
+ clusterClient
+ .sendRequest(
+ jobConfigHeaders,
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get(
+
operatorConfig.getFlinkClientTimeout().toSeconds(),
+ TimeUnit.SECONDS);
+ if (configurationInfo != null) {
+ if (configurationInfo.getExecutionConfigInfo() != null) {
+ jobConfig.put(
+ "parallelism.default",
+ String.valueOf(
+ configurationInfo
+ .getExecutionConfigInfo()
+ .getParallelism()));
+ jobConfig.put(
+ "pipeline.object-reuse",
Review Comment:
We should use the config options
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,85 @@ public boolean observe(FlinkResourceContext<R> ctx) {
return false;
}
+ /**
+ * Fetch runtime configuration from the Flink REST API and cache it in the
config manager when
+ * the job is in RUNNING state and no cached config exists for the current
job ID. After
+ * caching, the memoized observe config is cleared so subsequent consumers
see the merged
+ * values.
+ */
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ if (clusterJobStatus.getJobState() != JobStatus.RUNNING) {
+ return;
+ }
+
+ var resource = ctx.getResource();
+ var jobStatus = resource.getStatus().getJobStatus();
+ String namespace = resource.getMetadata().getNamespace();
+ String name = resource.getMetadata().getName();
+ String jobId = jobStatus.getJobId();
+
+ if (jobId == null) {
+ return;
+ }
Review Comment:
why would jobId be null?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
