Dennis-Mircea commented on code in PR #1000:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1000#discussion_r3088141833
##########
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:
I see no explicit reason to do this as the cache TTL will be hit anyway. The
existing `LoadingCache` for `observeConfig` doesn't do explicit invalidation
either (it relies on key-based natural misses + TTL), so the
`runtimeConfigCache` doesn't need it for consistency.
In this way, if we want to call `invalidateRuntimeConfig` in all valid
scenarios, we should also include `reconcileOtherChanges` as one of those cases.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,67 @@ 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();
+ var configManager = ctx.getConfigManager();
+
+ if (configManager.getRuntimeConfig(namespace, name,
jobId).isPresent()) {
+ LOG.debug("Runtime configuration already cached");
+ return;
+ }
+
+ LOG.debug("Fetching runtime configuration");
+ var conf = ctx.getObserveConfig();
+ var flinkService = ctx.getFlinkService();
+ var jobIdObj = JobID.fromHexString(jobId);
+ Map<String, String> runtimeConfig = new HashMap<>();
+ boolean fetchFailed = false;
+
+ try {
+ runtimeConfig.putAll(flinkService.getJobManagerConfiguration(conf,
jobIdObj));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch JobManager configuration", e);
+ }
+
+ try {
+ runtimeConfig.putAll(flinkService.getJobConfiguration(conf,
jobIdObj));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch job configuration", e);
+ }
+
+ try {
+
runtimeConfig.putAll(flinkService.getJobCheckpointConfiguration(conf,
jobIdObj));
+ } catch (Exception e) {
+ fetchFailed = true;
+ LOG.warn("Failed to fetch checkpoint configuration", e);
+ }
+
+ if (runtimeConfig.isEmpty() && fetchFailed) {
+ LOG.warn("All runtime config REST fetches failed, will retry next
cycle");
+ return;
+ }
+
+ configManager.putRuntimeConfig(namespace, name, jobId, runtimeConfig);
+ ctx.clearObserveConfig();
Review Comment:
Why to clear the `observeConfig` here? By having this in place it re-runs
the entire config generation pipeline (cache lookup, key building,
`ObjectMapper` serialization, cloning, spec config application) just to get an
identical base config cache when the `getObserveConfig()` will be called.
I'd say that it makes more sense to apply the running config support in here:
```suggestion
var currentObserveConfig = ctx.getObserveConfig();
if (currentObserveConfig != null) {
runtimeConfig.forEach(currentObserveConfig::setString);
}
```
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -88,6 +90,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;
Review Comment:
Using `jobId` as part of the cache key assumes that runtime configuration is
immutable for the lifetime of a job instance. This assumption may not hold for
in-place scaling (where parallelism changes without a new jobId) or future
dynamic configuration features such as
[FLIP-530](https://cwiki.apache.org/confluence/display/FLINK/FLIP-514%3A+Custom+Evaluator+plugin+for+Flink+Autoscaler)
and
[FLIP-571](https://cwiki.apache.org/confluence/display/FLINK/FLIP-571%3A+Support+Dynamically+Updating+Checkpoint+Configuration+at+Runtime+via+REST+API).
For the dynamic configuration we can adapt in the future (maybe with a
scheduled watcher in the same way as `scheduleConfigWatcher`), but for the
in-place scaling this is for sure a limiting factor and requires specific cache
invalidation.
The reason why the in-place scaling explicitly requires a cache invalidation
is because it changes the running Flink job's configuration (parallelism, task
slots), without changing the jobId. The current implementation as part of
`FlinkResourceContext#getObserveConfig` is placing the `runtimeConfig` with
higher precedence over `observeConfig`, meaning that it will outdate a new
scaled `observeConfig`.
To resolve this limitation my suggestion is to add the following code as
part at `AbstractFlinkResourceReconciler.java:367`:
```java
var resource = ctx.getResource();
ctx.getConfigManager()
.invalidateRuntimeConfig(
resource.getMetadata().getNamespace(),
resource.getMetadata().getName());
```
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java:
##########
@@ -111,6 +112,67 @@ 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();
Review Comment:
There are a lot of mixtures between concrete types and `var`. I'd suggest to
stick just to `var`.
--
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]
