Dennis-Mircea commented on code in PR #1000:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1000#discussion_r3093988329
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,46 @@ 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 runtime values are applied directly to the
already-memoized observe config so
+ * subsequent consumers see the merged values without a full config
regeneration.
+ */
+ private void fetchAndCacheRuntimeConfig(
+ FlinkResourceContext<R> ctx, JobStatusMessage clusterJobStatus) {
+ if (clusterJobStatus.getJobState() != JobStatus.RUNNING) {
Review Comment:
I think this condition restricts too much. I'd rather go with something like
this:
```java
if (clusterJobStatus.getJobState().isGloballyTerminalState()) {
return;
}
```
Or not even have at all this condition. @gyfora what do you say?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java:
##########
@@ -97,7 +97,27 @@ public Configuration getObserveConfig() {
if (observeConfig != null) {
return observeConfig;
}
- return observeConfig = createObserveConfig();
+ observeConfig = createObserveConfig();
+ if (observeConfig != null) {
+ configManager.applyCachedRuntimeConfig(getResource(),
observeConfig);
+ }
+ return observeConfig;
+ }
+
+ /**
+ * Clear the memoized observe config so it will be regenerated on next
access. This must be
+ * called after the observer updates the runtime configuration cache so
that subsequent
+ * consumers (other observers, reconciler) see the merged configuration.
+ */
+ public void clearObserveConfig() {
Review Comment:
This method is currently unused.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -873,6 +879,111 @@ public JobExceptionsInfoWithHistory getJobExceptions(
}
}
+ @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.warn("Failed to fetch JobManager configuration", e);
Review Comment:
I'll log these messages as ERROR not WARNING, since we encountered an
exception. Same for all the below cases.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java:
##########
@@ -111,10 +111,14 @@ public void deploy(
@Override
protected boolean cancelJob(FlinkResourceContext<FlinkSessionJob> ctx,
SuspendMode suspendMode)
throws Exception {
+ var resource = ctx.getResource();
+ ctx.getConfigManager()
+ .invalidateRuntimeConfig(
Review Comment:
By calling `invalidateRuntimeConfig` for each concrete case, I’ll focus even
more on adding tests to cover all of these scenarios, and by assuming that
@gyfora agrees this is the most suitable approach.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -351,6 +372,81 @@ public Configuration getObserveConfig(FlinkDeployment
deployment) {
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(
Review Comment:
Currently, within a single cycle, the cache is hit in:
1. `getObserveConfig()` -> `applyCachedRuntimeConfig()` ->
`getRuntimeConfig(ns, name, jobId)`
2. `fetchAndCacheRuntimeConfig()` -> `configManager.getRuntimeConfig(ns,
name, jobId).isPresent()`
So there are 2 hits per cycle at this point (and maybe in the future will be
even more) for this `runtimeConfig`. For that, I'd suggest to introduce a
`runtimeConfig` variable as part of `FlinkResourceContext` that will follow the
same pattern as `observeConfig`. Having a `runtimeConfig` variable per context
will reduce the cache hits to 1.
--
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]
