[
https://issues.apache.org/jira/browse/FLINK-41458?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Giannis Polyzos updated FLINK-41458:
------------------------------------
Description:
h3. Summary
The autoscaler caps a source's parallelism at its partition count only for
Kafka and Pulsar. For every other source the only available limit is a single
maximum parallelism for the whole job, which forces a trade-off between
limiting downstream operators and leaving source subtasks idle.
h3. Problem
Scaling a source beyond its number of partitions or splits adds no throughput:
the extra subtasks receive no work. FLINK-40389 restored the cap for Kafka and
Pulsar, whose partition counts the autoscaler detects from connector-specific
metrics. Other sources, such as Apache Fluss tables with a fixed number of
buckets or the Iceberg source discussed in FLINK-31400, provide no such signal.
The only remaining option, {{{}job.autoscaler.vertex.max-parallelism{}}},
applies to every vertex of the job.
h3. Impact
Jobs that read from other sources must choose between two configurations,
neither of them correct:
* *Cap the whole job at the source's partition count.* Downstream operators,
for example a lookup join, cannot scale beyond it even when they are the
bottleneck.
* *Leave the job uncapped.* Source subtasks above the partition count stay
idle while holding slots.
h3. Proposed Solution
Either of the following would address the problem:
# {*}Per-vertex maximum parallelism{*}, configurable by vertex ID or operator
name.
# *A connector-independent partition count* (preferred): a standard way for
any source to report its number of partitions or splits, used by the autoscaler
as that vertex's limit. This removes the Kafka/Pulsar special case and lets
other connectors take part without changes to the autoscaler.
The second option likely benefits from a short discussion on the dev mailing
list, which can follow this ticket.
h3. Environment
Flink Kubernetes Operator 1.16.1, Flink 2.3.0, Apache Fluss 1.0
was:
h3. Summary
The autoscaler caps a source's parallelism at its partition count only for
Kafka and Pulsar. For every other source the only available limit is a single
maximum parallelism for the whole job, which forces a trade-off between
limiting downstream operators and leaving source subtasks idle.
h3. Problem
Scaling a source beyond its number of partitions or splits adds no throughput:
the extra subtasks receive no work. FLINK-40389 restored the cap for Kafka and
Pulsar, whose partition counts the autoscaler detects from connector-specific
metrics. Other sources, such as Apache Fluss tables with a fixed number of
buckets or the Iceberg source discussed in FLINK-31400, provide no such signal.
The only remaining option, {{job.autoscaler.vertex.max-parallelism}}, applies
to every vertex of the job.
h3. Impact
Jobs that read from other sources must choose between two configurations,
neither of them correct:
* *Cap the whole job at the source's partition count.* Downstream operators,
for example a lookup join, cannot scale beyond it even when they are the
bottleneck.
* *Leave the job uncapped.* Source subtasks above the partition count stay idle
while holding slots.
h3. Proposed Solution
Either of the following would address the problem:
# *Per-vertex maximum parallelism*, configurable by vertex ID or operator name.
# *A connector-independent partition count* (preferred): a standard way for any
source to report its number of partitions or splits, used by the autoscaler as
that vertex's limit. This removes the Kafka/Pulsar special case and lets other
connectors take part without changes to the autoscaler.
The second option likely benefits from a short discussion on the dev mailing
list, which can follow this ticket.
h3. Environment
Flink Kubernetes Operator 1.16.1, Flink 2.3.0, Apache Fluss 1.0 sources with 9
buckets per table.
> Autoscaler: Support Max Parallelism For Sources Other Than Kafka And Pulsar
> ---------------------------------------------------------------------------
>
> Key: FLINK-41458
> URL: https://issues.apache.org/jira/browse/FLINK-41458
> Project: Flink
> Issue Type: Improvement
> Components: Kubernetes Operator
> Affects Versions: kubernetes-operator-1.16.1
> Reporter: Giannis Polyzos
> Priority: Major
>
> h3. Summary
> The autoscaler caps a source's parallelism at its partition count only for
> Kafka and Pulsar. For every other source the only available limit is a single
> maximum parallelism for the whole job, which forces a trade-off between
> limiting downstream operators and leaving source subtasks idle.
> h3. Problem
> Scaling a source beyond its number of partitions or splits adds no
> throughput: the extra subtasks receive no work. FLINK-40389 restored the cap
> for Kafka and Pulsar, whose partition counts the autoscaler detects from
> connector-specific metrics. Other sources, such as Apache Fluss tables with a
> fixed number of buckets or the Iceberg source discussed in FLINK-31400,
> provide no such signal.
> The only remaining option, {{{}job.autoscaler.vertex.max-parallelism{}}},
> applies to every vertex of the job.
> h3. Impact
> Jobs that read from other sources must choose between two configurations,
> neither of them correct:
> * *Cap the whole job at the source's partition count.* Downstream operators,
> for example a lookup join, cannot scale beyond it even when they are the
> bottleneck.
> * *Leave the job uncapped.* Source subtasks above the partition count stay
> idle while holding slots.
> h3. Proposed Solution
> Either of the following would address the problem:
> # {*}Per-vertex maximum parallelism{*}, configurable by vertex ID or
> operator name.
> # *A connector-independent partition count* (preferred): a standard way for
> any source to report its number of partitions or splits, used by the
> autoscaler as that vertex's limit. This removes the Kafka/Pulsar special case
> and lets other connectors take part without changes to the autoscaler.
> The second option likely benefits from a short discussion on the dev mailing
> list, which can follow this ticket.
> h3. Environment
> Flink Kubernetes Operator 1.16.1, Flink 2.3.0, Apache Fluss 1.0
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
