[
https://issues.apache.org/jira/browse/FLINK-40474?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40474:
-----------------------------------
Labels: pull-request-available (was: )
> Autoscaler parallelism overrides serialized in the operator's YAML dialect
> break job submission for Flink 1.x jobs
> ------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40474
> URL: https://issues.apache.org/jira/browse/FLINK-40474
> Project: Flink
> Issue Type: Improvement
> Components: Autoscaler, Kubernetes Operator
> Affects Versions: kubernetes-operator-1.16.0
> Reporter: Dennis-Mircea Ciupitu
> Priority: Critical
> Labels: pull-request-available
> Fix For: kubernetes-operator-1.16.0
>
>
> h1. Problem
> After the autoscaler scales a job, the next full job submission fails, and
> the resource ends up in a failed state:
> * *FlinkDeployment* (application mode): the first spec upgrade, last-state
> restore or any other full redeploy after a scaling fails at submission.
> * {*}FlinkSessionJob{*}: every submit after a scaling fails. Session jobs
> are re-submitted on each upgrade, so this triggers even sooner.
> The job's JobManager rejects the configuration written by the operator:
> {code:java}
> Could not parse value '{90bea66de1c231edf33913ecd54406c1: '3',
> cbc357ccb763df2852fee8c4fc7d55f2: '2'}'
> for key 'pipeline.jobvertex-parallelism-overrides'. -> Could not split
> string. Illegal quoting at position: 36
> {code}
> h1. Root cause
> Currently, Flink has two textual dialects for Map-typed config values:
> {noformat}
> legacy (flink-conf.yaml) -> vertexA:2,vertexB:1
> standard (config.yaml) -> {vertexA: '2', vertexB: '1'}
> {noformat}
> The problem in the new 1.16.0 is that the operator boots up with the standard
> YAML parser loaded (added by FLINK-39791), which is done via
> {{{}GlobalConfiguration.loadConfiguration(){}}}, meaning that for the full
> lifecycle of the operator, for all the subsequent reconciliation cycles, the
> standard YAML parser will be used against all the submitted deployments,
> which can run on versions lower than 2.0 (which are using the legacy YAML
> parser), or equal and higher than 2.0 (which are using the standard YAML
> parser).
> By default, the standard YAML parser can fallback to the legacy YAML parser,
> which is the reason why everything was working before, because the legacy
> YAML format still represents a general format that both parsers understand.
> However, if a legacy YAML parser receives a standard YAML format, this will
> translate into an error.
> This is exactly what happens with the autoscaler: when a scaling decision is
> realized, {{KubernetesScalingRealizer#getOverrideString}} serializes the
> parallelism overrides through {{ConfigurationUtils.convertValue(map,
> String.class)}}, which internally follows the parser loaded at boot time (via
> {{GlobalConfiguration.isStandardYaml()}}), so on 1.16.0 it produces the
> standard YAML format. The resulting string is persisted into the resource's
> {{spec.flinkConfiguration}}, where it outlives the reconciliation cycle, and
> it is only parsed again later, on the next full job submission, inside the
> deployment's own JVM.
> Note that the realizer code itself did not change (it is byte-identical to
> 1.15), the defect was latent and became visible once FLINK-39791 changed the
> parser the operator boots with. FLINK-40055 already solved the same class of
> problem for typed config values, by re-serializing them into the target
> deployment's dialect at the deploy boundary
> ({{{}configToMapWithVersionDialect{}}}), but a value that was already turned
> into a string before reaching that boundary passes through unchanged, both
> when the deployment's config file is written and when the session job jar-run
> request body is built.
> For completeness, a second producer of such strings exists in
> {{{}FlinkConfigManager{}}}: Map or List typed job options placed in the
> operator's defaultConfiguration are stringified with the operator's parser
> (through {{{}Configuration#toMap{}}}) before reaching the same boundary, with
> the same effect for deployments running on versions lower than 2.0.
> h1. Affected versions
> This problem is in place only for the new 1.16.0 operator release candidate,
> as it contains the new standard YAML parser support. The previous versions
> are unaffected.
> h1. Reproduction
> The following steps can be followed for a full reproduction:
> # Install the 1.16 operator with default Helm values, so the operator runs
> on {{{}config.yaml{}}}.
> # Deploy a job with {{job.autoscaler.enabled}} set to true and let it
> perform one scaling. The CR will contain
> {{pipeline.jobvertex-parallelism-overrides}} in the standard flow form.
> # Trigger any full redeploy (spec change, {{{}restartNonce{}}}) for a Flink
> 1.x job.
> # Submission fails with the parse error above.
> This was reproduced against release-1.16.0-rc2 for both application mode
> (Flink 1.20 job) and a session job with a Flink 1.20 session cluster.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
