gyfora commented on code in PR #1000:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1000#discussion_r3044055592
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -352,8 +393,193 @@ public Configuration getObserveConfig(FlinkDeployment
deployment) {
return conf;
}
+ /**
+ * Get the observe configuration enhanced with runtime configuration
overrides fetched from the
+ * job. This method fetches the actual runtime configuration from the job
via REST API and
+ * overrides the deployment spec configuration with the actual values.
+ *
+ * @param deployment Deployment resource
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @return Observe config enhanced with runtime configuration overrides
+ */
+ public Configuration getObserveConfigWithRuntimeOverrides(
+ FlinkDeployment deployment, FlinkService flinkService) {
+ // Get the base observe configuration
+ Configuration conf = getObserveConfig(deployment);
+
+ // Apply runtime configuration overrides if job is running
+ applyRuntimeConfigurationOverrides(deployment, flinkService, conf);
+
+ return conf;
+ }
+
+ /**
+ * Get configuration for interacting with session jobs enhanced with
runtime configuration
+ * overrides. This method fetches the actual runtime configuration from
the job via REST API and
+ * overrides the session job spec configuration with the actual values.
+ *
+ * @param sessionJob Session job resource
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @return Session job config enhanced with runtime configuration overrides
+ */
+ public Configuration getSessionJobConfigWithRuntimeOverrides(
+ FlinkSessionJob sessionJob, FlinkService flinkService,
Configuration conf) {
+ // Apply runtime configuration overrides if job is running
+ applyRuntimeConfigurationOverrides(sessionJob, flinkService, conf);
+
+ return conf;
+ }
+
+ /**
+ * Generic method to apply runtime configuration overrides from any Flink
resource to the
+ * provided configuration. This method fetches the actual runtime
configuration from the running
+ * job and overrides the resource spec configuration with the actual
values to ensure the
+ * operator works with the correct configuration values.
+ *
+ * <p>Uses a job-state-aware cache to ensure runtime configuration is only
fetched once per job
+ * when it reaches RUNNING state.
+ *
+ * @param resource Flink resource (FlinkDeployment or FlinkSessionJob)
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @param conf Configuration to apply overrides to
+ */
+ private <T extends AbstractFlinkResource<?, ?>> void
applyRuntimeConfigurationOverrides(
+ T resource, FlinkService flinkService, Configuration conf) {
+ try {
+ // Check if job is running and has a job ID
+ var jobStatus = resource.getStatus().getJobStatus();
+ if (jobStatus == null || jobStatus.getJobId() == null) {
+ LOG.debug("No job ID available for runtime configuration
fetch");
+ return;
+ }
+
+ JobID jobId = JobID.fromHexString(jobStatus.getJobId());
+ JobStatus jobState = jobStatus.getState();
+
+ // Only fetch runtime config if job is in a running state
+ if (jobState == null || !jobState.equals(JobStatus.RUNNING)) {
+ LOG.debug(
+ "Job {} is not in RUNNING state (current: {}),
skipping runtime config fetch",
+ jobId,
+ jobState);
+ return;
+ }
Review Comment:
Please see my other comment relating the observer logic, but overall once
the runtime config is cached for the jobid we should apply it, even if the job
is failing/restarting/etc. The configmanager doesn't need this logic after that
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -352,8 +393,193 @@ public Configuration getObserveConfig(FlinkDeployment
deployment) {
return conf;
}
+ /**
+ * Get the observe configuration enhanced with runtime configuration
overrides fetched from the
+ * job. This method fetches the actual runtime configuration from the job
via REST API and
+ * overrides the deployment spec configuration with the actual values.
+ *
+ * @param deployment Deployment resource
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @return Observe config enhanced with runtime configuration overrides
+ */
+ public Configuration getObserveConfigWithRuntimeOverrides(
+ FlinkDeployment deployment, FlinkService flinkService) {
+ // Get the base observe configuration
+ Configuration conf = getObserveConfig(deployment);
+
+ // Apply runtime configuration overrides if job is running
+ applyRuntimeConfigurationOverrides(deployment, flinkService, conf);
+
+ return conf;
+ }
+
+ /**
+ * Get configuration for interacting with session jobs enhanced with
runtime configuration
+ * overrides. This method fetches the actual runtime configuration from
the job via REST API and
+ * overrides the session job spec configuration with the actual values.
+ *
+ * @param sessionJob Session job resource
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @return Session job config enhanced with runtime configuration overrides
+ */
+ public Configuration getSessionJobConfigWithRuntimeOverrides(
+ FlinkSessionJob sessionJob, FlinkService flinkService,
Configuration conf) {
+ // Apply runtime configuration overrides if job is running
+ applyRuntimeConfigurationOverrides(sessionJob, flinkService, conf);
+
+ return conf;
+ }
+
+ /**
+ * Generic method to apply runtime configuration overrides from any Flink
resource to the
+ * provided configuration. This method fetches the actual runtime
configuration from the running
+ * job and overrides the resource spec configuration with the actual
values to ensure the
+ * operator works with the correct configuration values.
+ *
+ * <p>Uses a job-state-aware cache to ensure runtime configuration is only
fetched once per job
+ * when it reaches RUNNING state.
+ *
+ * @param resource Flink resource (FlinkDeployment or FlinkSessionJob)
+ * @param flinkService FlinkService to use for fetching runtime
configuration
+ * @param conf Configuration to apply overrides to
+ */
+ private <T extends AbstractFlinkResource<?, ?>> void
applyRuntimeConfigurationOverrides(
+ T resource, FlinkService flinkService, Configuration conf) {
+ try {
+ // Check if job is running and has a job ID
+ var jobStatus = resource.getStatus().getJobStatus();
+ if (jobStatus == null || jobStatus.getJobId() == null) {
+ LOG.debug("No job ID available for runtime configuration
fetch");
+ return;
+ }
+
+ JobID jobId = JobID.fromHexString(jobStatus.getJobId());
+ JobStatus jobState = jobStatus.getState();
+
+ // Only fetch runtime config if job is in a running state
+ if (jobState == null || !jobState.equals(JobStatus.RUNNING)) {
+ LOG.debug(
+ "Job {} is not in RUNNING state (current: {}),
skipping runtime config fetch",
+ jobId,
+ jobState);
+ return;
+ }
+
+ // Check if JobManager is ready (only for FlinkDeployment)
+ if (resource instanceof FlinkDeployment) {
+ FlinkDeployment deployment = (FlinkDeployment) resource;
+ if (deployment.getStatus().getJobManagerDeploymentStatus()
+ != JobManagerDeploymentStatus.READY) {
+ LOG.debug(
+ "JobManager is not ready, skipping runtime config
fetch for job {}",
+ jobId);
+ return;
+ }
Review Comment:
In general I think it's not the right approach to have this logic here. We
should move this to the observers where job state is tracked already and we
know exactly when the rest api can be accessed.
So instead of accessing the rest api here, expose methods for
setting/invalidating/getting the runtime config in the ConfigManager. Then the
observer could check if the configmanager has a cached config already for the
current job id , and if not update it. Same way, the reconciler could
invalidate runtime configs after an cancel/upgrade.
--
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]
