[
https://issues.apache.org/jira/browse/FLINK-40474?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dennis-Mircea Ciupitu updated FLINK-40474:
------------------------------------------
Description:
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.
was:
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.
> 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
> 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)
