[jira] [Updated] (FLINK-40055) Operator serializes managed-deployment config in its own YAML dialect, breaking Flink 1.x deployments when the operator itself runs on standard config.yaml

["Gerk Elznik (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40055?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Gerk Elznik updated FLINK-40055:
--------------------------------
    Description: 
When the operator's own configuration is loaded from a standard-mode {{config.yaml}} (rather than legacy {{{}flink-conf.yaml{}}}), the config files it generates for *managed Flink 1.x deployments* contain standard-YAML-typed values that Flink 1.x's legacy parser cannot read. An application-mode deployment then crashloops at JobManager startup:
{code:java}
ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Could not create application program.
java.lang.RuntimeException: java.net.URISyntaxException: Illegal character in scheme name at index 0: ['local:///opt/flink/usrlib/myjob.jar']
  at org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint.fetchArtifacts(...) [flink-dist-1.20.4.jar]
{code}
h3. Reproduction (A/B, only variable = the operator's own config file format)

Same operator image (current main), same kind cluster, same {{v1_20}} application-mode FlinkDeployment ({{{}local://{}}} StateMachineExample):
|| ||operator reads {{config.yaml}}||operator reads {{flink-conf.yaml}} (today's chart default)||
|operator config mode|standard YAML|legacy|
|value written into the deployment's {{flink-conf.yaml}}|{{pipeline.jars: ['local:///…/myjob.jar']}}|{{pipeline.jars: local:///…/myjob.jar}}|
|{{v1_20}} JobManager|URISyntaxException → CrashLoopBackOff|READY / RUNNING|

Today this is reachable e.g. via a conf-override dir or a custom mount that places a {{config.yaml}} in the operator's conf directory ({{{}GlobalConfiguration.loadConfiguration{}}} prefers it). FLINK-39791 (PR #1126) would make it the default path by mounting the operator config as {{{}config.yaml{}}}, which is how it was found — see the analysis on [https://github.com/apache/flink-kubernetes-operator/pull/1126].
h3. Root cause

The YAML dialect used for serialization is a *process-global* property in flink-core: when the operator loads its own configuration from {{{}config.yaml{}}}, {{GlobalConfiguration.isStandardYaml()}} becomes true JVM-wide, and every no-arg {{new Configuration()}} — including {{{}Configuration#clone(){}}}, which is implemented as {{new Configuration() + addAll}} — creates standard-dialect configs from then on. The managed deployment's effective config therefore speaks the *operator's* dialect instead of the {*}target deployment's{*}:
 # The deploy config handed to the reconciler ({{{}FlinkConfigManager#getConfig{}}} → {{{}cache.get(key).clone(){}}}) carries {{standardYaml=true}} (via the process-global flag; the cached original also inherits it through {{new Configuration(defaultConfig)}} in {{{}getDefaultConfig{}}}).
 # {{AbstractFlinkService#removeOperatorConfigs}} round-trips that config through {{config.toMap()}} + {{{}setString{}}}. {{Configuration#toMap}} serializes every value with {{{}ConfigurationUtils.convertToString(value, standardYaml){}}}; with {{standardYaml=true}} a {{List}} (e.g. {{{}PipelineOptions.JARS{}}}, set as a typed list by {{{}FlinkConfigBuilder{}}}) becomes the string {{{}"['local:///…']"{}}}.
 # {{FlinkConfMountDecorator#getClusterSideConfData}} then correctly builds {{new Configuration(useStandardYamlConfig())}} — false for v1_x, so the file is named {{flink-conf.yaml}} and written legacy-style — but the damage is done: the bracketed *string* is written verbatim. The comment there ("would always be false currently", from FLINK-37236) documents exactly the assumption that no longer holds once the operator itself runs on standard config.
 # Flink 1.x's legacy parser reads {{['local://…']}} as a raw string → {{{}URISyntaxException{}}}.

The chain was reproduced in isolation against flink-core 1.20.4: after {{GlobalConfiguration.loadConfiguration}} of a directory containing {{{}config.yaml{}}}, the clone→{{{}toMap{}}}→legacy-write sequence emits {{{}pipeline.jars: ['local:///…']{}}}, while a dialect-aware copy emits the legacy-parseable {{{}pipeline.jars: local:///…{}}}.

Affects any list-typed option (not just {{{}pipeline.jars{}}}) for any {{v1_xx}} deployment. Session-job submission has the same dialect leak: {{AbstractFlinkService#submitJobToSessionCluster}} sends {{conf.toMap()}} (operator-dialect strings) in the {{JarRunRequestBody}} to v1_17+ session clusters.
h3. Proposed fix

Fix at the {*}serialization boundaries{*}, leaving in-process parse semantics untouched (stamping the target dialect onto the long-lived resource configs was prototyped and rejected: {{standardYaml}} also governs how raw string values are {*}parsed{*}, so it changes read behavior for every consumer — and any downstream {{{}new Configuration(){}}}/{{{}clone(){}}} silently resets it anyway):
 * {{AbstractFlinkService#removeOperatorConfigs}} — copy via the copy constructor and {{removeKey}} directly instead of the {{{}toMap(){}}}/{{{}setString{}}} round-trip, so typed values (e.g. the {{pipeline.jars}} list) survive untouched to the boundaries that already render per target version ({{{}FlinkConfMountDecorator{}}});
 * {{AbstractFlinkService#runJar}} — serialize the session-job REST config map in the receiving cluster's dialect (a throwaway dialect-stamped copy just for the {{{}JarRunRequestBody{}}});
 * the Flink-2.0 dialect threshold becomes one shared authority: {{FlinkConfMountDecorator#useStandardYamlConfig(FlinkVersion)}} (public static; the instance method delegates).

Verified with a from-source operator build across all four operator-format × deployment-version combinations: the {{v1_20}} application deployment that crashloops without the fix reaches RUNNING with it (operator on {{{}config.yaml{}}}), Flink 2.x deployments work under both operator formats, and the generated deployment config is identical regardless of the operator's own format (only per-run identity values differ). Also exercised beyond CI with an in-place operator config-format migration (helm chart flip legacy↔standard under running v1 and v2 jobs, JobManager pods killed after each flip to force a config re-read): no job disruption, no pod churn, generated configs byte-identical. Unit test included ({{{}AbstractFlinkServiceTest{}}}) that pins the process-global standard-yaml flag and asserts typed-value survival plus per-version serialization.

This is a prerequisite for FLINK-39791 (chart mounting the operator config as {{{}config.yaml{}}}).
 # 
 ## Issue links to add after creation

 - blocks: FLINK-39791
 - relates to: FLINK-37236


  was:
When the operator's own configuration is loaded from a standard-mode {{config.yaml}} (rather than legacy {{{}flink-conf.yaml{}}}), the config files it generates for *managed Flink 1.x deployments* contain standard-YAML-typed values that Flink 1.x's legacy parser cannot read. An application-mode deployment then crashloops at JobManager startup:
{code:java}
ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Could not create application program.
java.lang.RuntimeException: java.net.URISyntaxException: Illegal character in scheme name at index 0: ['local:///opt/flink/usrlib/myjob.jar']
  at org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint.fetchArtifacts(...) [flink-dist-1.20.4.jar]
{code}
h3. Reproduction (A/B, only variable = the operator's own config file format)

Same operator image (current main), same kind cluster, same {{v1_20}} application-mode FlinkDeployment ({{{}local://{}}} StateMachineExample):
|| ||operator reads {{config.yaml}}||operator reads {{flink-conf.yaml}} (today's chart default)||
|operator config mode|standard YAML|legacy|
|value written into the deployment's {{flink-conf.yaml}}|{{pipeline.jars: ['local:///…/myjob.jar']}}|{{pipeline.jars: local:///…/myjob.jar}}|
|{{v1_20}} JobManager|URISyntaxException → CrashLoopBackOff|READY / RUNNING|

Today this is reachable e.g. via a conf-override dir or a custom mount that places a {{config.yaml}} in the operator's conf directory ({{{}GlobalConfiguration.loadConfiguration{}}} prefers it). FLINK-39791 (PR #1126) would make it the default path by mounting the operator config as {{{}config.yaml{}}}, which is how it was found — see the analysis on [https://github.com/apache/flink-kubernetes-operator/pull/1126].
h3. Root cause

The YAML dialect used for serialization is a *process-global* property in flink-core: when the operator loads its own configuration from {{{}config.yaml{}}}, {{GlobalConfiguration.isStandardYaml()}} becomes true JVM-wide, and every no-arg {{new Configuration()}} — including {{{}Configuration#clone(){}}}, which is implemented as {{new Configuration() + addAll}} — creates standard-dialect configs from then on. The managed deployment's effective config therefore speaks the *operator's* dialect instead of the {*}target deployment's{*}:
 # The deploy config handed to the reconciler ({{{}FlinkConfigManager#getConfig{}}} → {{{}cache.get(key).clone(){}}}) carries {{standardYaml=true}} (via the process-global flag; the cached original also inherits it through {{new Configuration(defaultConfig)}} in {{{}getDefaultConfig{}}}).
 # {{AbstractFlinkService#removeOperatorConfigs}} round-trips that config through {{config.toMap()}} + {{{}setString{}}}. {{Configuration#toMap}} serializes every value with {{{}ConfigurationUtils.convertToString(value, standardYaml){}}}; with {{standardYaml=true}} a {{List}} (e.g. {{{}PipelineOptions.JARS{}}}, set as a typed list by {{{}FlinkConfigBuilder{}}}) becomes the string {{{}"['local:///…']"{}}}.
 # {{FlinkConfMountDecorator#getClusterSideConfData}} then correctly builds {{new Configuration(useStandardYamlConfig())}} — false for v1_x, so the file is named {{flink-conf.yaml}} and written legacy-style — but the damage is done: the bracketed *string* is written verbatim. The comment there ("would always be false currently", from FLINK-37236) documents exactly the assumption that no longer holds once the operator itself runs on standard config.
 # Flink 1.x's legacy parser reads {{['local://…']}} as a raw string → {{{}URISyntaxException{}}}.

The chain was reproduced in isolation against flink-core 1.20.4: after {{GlobalConfiguration.loadConfiguration}} of a directory containing {{{}config.yaml{}}}, the clone→{{{}toMap{}}}→legacy-write sequence emits {{{}pipeline.jars: ['local:///…']{}}}, while a dialect-aware copy emits the legacy-parseable {{{}pipeline.jars: local:///…{}}}.

Affects any list-typed option (not just {{{}pipeline.jars{}}}) for any {{v1_xx}} deployment. Session-job submission has the same dialect leak: {{AbstractFlinkService#submitJobToSessionCluster}} sends {{conf.toMap()}} (operator-dialect strings) in the {{JarRunRequestBody}} to v1_17+ session clusters.
h3. Proposed fix

Fix at the {*}serialization boundaries{*}, leaving in-process parse semantics untouched (stamping the target dialect onto the long-lived resource configs was prototyped and rejected: {{standardYaml}} also governs how raw string values are {*}parsed{*}, so it changes read behavior for every consumer — and any downstream {{{}new Configuration(){}}}/{{{}clone(){}}} silently resets it anyway):
 * {{AbstractFlinkService#removeOperatorConfigs}} — copy via the copy constructor and {{removeKey}} directly instead of the {{{}toMap(){}}}/{{{}setString{}}} round-trip, so typed values (e.g. the {{pipeline.jars}} list) survive untouched to the boundaries that already render per target version ({{{}FlinkConfMountDecorator{}}});
 * {{AbstractFlinkService#runJar}} — serialize the session-job REST config map in the receiving cluster's dialect (a throwaway dialect-stamped copy just for the {{{}JarRunRequestBody{}}});
 * the Flink-2.0 dialect threshold becomes one shared authority: {{FlinkConfMountDecorator#useStandardYamlConfig(FlinkVersion)}} (public static; the instance method delegates).

Verified with a from-source operator build across all four operator-format × deployment-version combinations: the {{v1_20}} application deployment that crashloops without the fix reaches RUNNING with it (operator on {{{}config.yaml{}}}), Flink 2.x deployments work under both operator formats, and the generated deployment config is identical regardless of the operator's own format (only per-run identity values differ). Also exercised beyond CI with an in-place operator config-format migration (helm chart flip legacy↔standard under running v1 and v2 jobs, JobManager pods killed after each flip to force a config re-read): no job disruption, no pod churn, generated configs byte-identical. Unit test included ({{{}AbstractFlinkServiceTest{}}}) that pins the process-global standard-yaml flag and asserts typed-value survival plus per-version serialization.

This is a prerequisite for FLINK-39791 (chart mounting the operator config as {{{}config.yaml{}}}).
 # 
 ## Issue links to add after creation

 - *{*}blocks{*}* FLINK-39791
 - *{*}relates to{*}* FLINK-37236


> Operator serializes managed-deployment config in its own YAML dialect, breaking Flink 1.x deployments when the operator itself runs on standard config.yaml
> -----------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-40055
>                 URL: https://issues.apache.org/jira/browse/FLINK-40055
>             Project: Flink
>          Issue Type: Bug
>          Components: Kubernetes Operator
>    Affects Versions: kubernetes-operator-1.15.0
>            Reporter: Gerk Elznik
>            Priority: Major
>              Labels: operator
>
> When the operator's own configuration is loaded from a standard-mode {{config.yaml}} (rather than legacy {{{}flink-conf.yaml{}}}), the config files it generates for *managed Flink 1.x deployments* contain standard-YAML-typed values that Flink 1.x's legacy parser cannot read. An application-mode deployment then crashloops at JobManager startup:
> {code:java}
> ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Could not create application program.
> java.lang.RuntimeException: java.net.URISyntaxException: Illegal character in scheme name at index 0: ['local:///opt/flink/usrlib/myjob.jar']
>   at org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint.fetchArtifacts(...) [flink-dist-1.20.4.jar]
> {code}
> h3. Reproduction (A/B, only variable = the operator's own config file format)
> Same operator image (current main), same kind cluster, same {{v1_20}} application-mode FlinkDeployment ({{{}local://{}}} StateMachineExample):
> || ||operator reads {{config.yaml}}||operator reads {{flink-conf.yaml}} (today's chart default)||
> |operator config mode|standard YAML|legacy|
> |value written into the deployment's {{flink-conf.yaml}}|{{pipeline.jars: ['local:///…/myjob.jar']}}|{{pipeline.jars: local:///…/myjob.jar}}|
> |{{v1_20}} JobManager|URISyntaxException → CrashLoopBackOff|READY / RUNNING|
> Today this is reachable e.g. via a conf-override dir or a custom mount that places a {{config.yaml}} in the operator's conf directory ({{{}GlobalConfiguration.loadConfiguration{}}} prefers it). FLINK-39791 (PR #1126) would make it the default path by mounting the operator config as {{{}config.yaml{}}}, which is how it was found — see the analysis on [https://github.com/apache/flink-kubernetes-operator/pull/1126].
> h3. Root cause
> The YAML dialect used for serialization is a *process-global* property in flink-core: when the operator loads its own configuration from {{{}config.yaml{}}}, {{GlobalConfiguration.isStandardYaml()}} becomes true JVM-wide, and every no-arg {{new Configuration()}} — including {{{}Configuration#clone(){}}}, which is implemented as {{new Configuration() + addAll}} — creates standard-dialect configs from then on. The managed deployment's effective config therefore speaks the *operator's* dialect instead of the {*}target deployment's{*}:
>  # The deploy config handed to the reconciler ({{{}FlinkConfigManager#getConfig{}}} → {{{}cache.get(key).clone(){}}}) carries {{standardYaml=true}} (via the process-global flag; the cached original also inherits it through {{new Configuration(defaultConfig)}} in {{{}getDefaultConfig{}}}).
>  # {{AbstractFlinkService#removeOperatorConfigs}} round-trips that config through {{config.toMap()}} + {{{}setString{}}}. {{Configuration#toMap}} serializes every value with {{{}ConfigurationUtils.convertToString(value, standardYaml){}}}; with {{standardYaml=true}} a {{List}} (e.g. {{{}PipelineOptions.JARS{}}}, set as a typed list by {{{}FlinkConfigBuilder{}}}) becomes the string {{{}"['local:///…']"{}}}.
>  # {{FlinkConfMountDecorator#getClusterSideConfData}} then correctly builds {{new Configuration(useStandardYamlConfig())}} — false for v1_x, so the file is named {{flink-conf.yaml}} and written legacy-style — but the damage is done: the bracketed *string* is written verbatim. The comment there ("would always be false currently", from FLINK-37236) documents exactly the assumption that no longer holds once the operator itself runs on standard config.
>  # Flink 1.x's legacy parser reads {{['local://…']}} as a raw string → {{{}URISyntaxException{}}}.
> The chain was reproduced in isolation against flink-core 1.20.4: after {{GlobalConfiguration.loadConfiguration}} of a directory containing {{{}config.yaml{}}}, the clone→{{{}toMap{}}}→legacy-write sequence emits {{{}pipeline.jars: ['local:///…']{}}}, while a dialect-aware copy emits the legacy-parseable {{{}pipeline.jars: local:///…{}}}.
> Affects any list-typed option (not just {{{}pipeline.jars{}}}) for any {{v1_xx}} deployment. Session-job submission has the same dialect leak: {{AbstractFlinkService#submitJobToSessionCluster}} sends {{conf.toMap()}} (operator-dialect strings) in the {{JarRunRequestBody}} to v1_17+ session clusters.
> h3. Proposed fix
> Fix at the {*}serialization boundaries{*}, leaving in-process parse semantics untouched (stamping the target dialect onto the long-lived resource configs was prototyped and rejected: {{standardYaml}} also governs how raw string values are {*}parsed{*}, so it changes read behavior for every consumer — and any downstream {{{}new Configuration(){}}}/{{{}clone(){}}} silently resets it anyway):
>  * {{AbstractFlinkService#removeOperatorConfigs}} — copy via the copy constructor and {{removeKey}} directly instead of the {{{}toMap(){}}}/{{{}setString{}}} round-trip, so typed values (e.g. the {{pipeline.jars}} list) survive untouched to the boundaries that already render per target version ({{{}FlinkConfMountDecorator{}}});
>  * {{AbstractFlinkService#runJar}} — serialize the session-job REST config map in the receiving cluster's dialect (a throwaway dialect-stamped copy just for the {{{}JarRunRequestBody{}}});
>  * the Flink-2.0 dialect threshold becomes one shared authority: {{FlinkConfMountDecorator#useStandardYamlConfig(FlinkVersion)}} (public static; the instance method delegates).
> Verified with a from-source operator build across all four operator-format × deployment-version combinations: the {{v1_20}} application deployment that crashloops without the fix reaches RUNNING with it (operator on {{{}config.yaml{}}}), Flink 2.x deployments work under both operator formats, and the generated deployment config is identical regardless of the operator's own format (only per-run identity values differ). Also exercised beyond CI with an in-place operator config-format migration (helm chart flip legacy↔standard under running v1 and v2 jobs, JobManager pods killed after each flip to force a config re-read): no job disruption, no pod churn, generated configs byte-identical. Unit test included ({{{}AbstractFlinkServiceTest{}}}) that pins the process-global standard-yaml flag and asserts typed-value survival plus per-version serialization.
> This is a prerequisite for FLINK-39791 (chart mounting the operator config as {{{}config.yaml{}}}).
>  # 
>  ## Issue links to add after creation
>  - blocks: FLINK-39791
>  - relates to: FLINK-37236



--
This message was sent by Atlassian Jira
(v8.20.10#820010)