You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Renxiang Zhou (Jira)" <ji...@apache.org> on 2025/07/22 09:46:00 UTC
[jira] [Updated] (FLINK-38131) Repartition BroadcastState with finished operators
[ https://issues.apache.org/jira/browse/FLINK-38131?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Renxiang Zhou updated FLINK-38131:
----------------------------------
Description:
FLINK-23475 proposes how to distribute Broadcast state to new operator instances to avoid new instances getting empty Broadcast state when there are some finished operator in the DAG, and the task is restarted without changing the parallelism. However, the current implementation seems to have some problems in determining whether there is a finished operator.
As the pic shown below, `isPartiallyReported` compares the value between BitSet's
`cardinality` and `size`, where `cardinality` represents the number of operators that contain the Broadcast state, but `size` just returns the number of bits of space actually in use by this BitSet. It makes no sense to compare these two parameters, which causes the Broadcast state of operators without finished status to be redistributed when the parallelism does not change, which is unnecessary. I think we should use the `parallelism` in the constructor instead of `reportedSubtaskIndices.size()`
!image-2025-07-22-17-34-22-479.png|width=528,height=334!
was:
FLINK-23475 proposes how to distribute Broadcast state to new operator instances to avoid new instances getting empty Broadcast state when there are some finished operator in the DAG, and the task is restarted without changing the parallelism. However, the current implementation seems to have some problems in determining whether there is a finished operator.
As the pic shown below, `isPartiallyReported` compares the value between BitSet's
`cardinality` and `size`, where `cardinality` represents the number of operators that contain the Broadcast state, but `size` just returns the number of bits of space actually in use by this BitSet. It makes no sense to compare these two parameters, which causes the Broadcast state of operators without finished status to be redistributed when the parallelism does not change, which is unnecessary. I think we should use the `parallelism` in the constructor instead of `reportedSubtaskIndices.size()`
!image-2025-07-22-17-34-22-479.png!
> Repartition BroadcastState with finished operators
> --------------------------------------------------
>
> Key: FLINK-38131
> URL: https://issues.apache.org/jira/browse/FLINK-38131
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
> Affects Versions: 1.14.0
> Reporter: Renxiang Zhou
> Priority: Major
> Fix For: 2.1.0
>
> Attachments: image-2025-07-22-17-34-22-479.png
>
>
> FLINK-23475 proposes how to distribute Broadcast state to new operator instances to avoid new instances getting empty Broadcast state when there are some finished operator in the DAG, and the task is restarted without changing the parallelism. However, the current implementation seems to have some problems in determining whether there is a finished operator.
> As the pic shown below, `isPartiallyReported` compares the value between BitSet's
> `cardinality` and `size`, where `cardinality` represents the number of operators that contain the Broadcast state, but `size` just returns the number of bits of space actually in use by this BitSet. It makes no sense to compare these two parameters, which causes the Broadcast state of operators without finished status to be redistributed when the parallelism does not change, which is unnecessary. I think we should use the `parallelism` in the constructor instead of `reportedSubtaskIndices.size()`
> !image-2025-07-22-17-34-22-479.png|width=528,height=334!
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)