You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "1996fanrui (via GitHub)" <gi...@apache.org> on 2026/10/21 14:23:34 UTC
[PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
1996fanrui opened a new pull request, #29549:
URL: https://github.com/apache/flink/pull/29549
Backporting https://github.com/apache/flink/pull/29374 and https://github.com/apache/flink/pull/29272 to 2.2
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #29549:
URL: https://github.com/apache/flink/pull/29549#issuecomment-6447287561
Thanks @1996fanrui
LGTM +1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "1996fanrui (via GitHub) via issues" <is...@flink.apache.org>.
1996fanrui commented on code in PR #29549:
URL: https://github.com/apache/flink/pull/29549#discussion_r4435352170
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -1782,13 +1785,64 @@ List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters
private static void replaceForwardPartitionerIfConsumerParallelismDoesNotMatch(
Environment environment, NonChainedOutput streamOutput, int outputIndex) {
- if (streamOutput.getPartitioner() instanceof ForwardPartitioner
- && environment.getWriter(outputIndex).getNumberOfSubpartitions()
- != environment.getTaskInfo().getNumberOfParallelSubtasks()) {
- LOG.debug(
- "Replacing forward partitioner with rebalance for {}",
- environment.getTaskInfo().getTaskNameWithSubtasks());
- streamOutput.setPartitioner(new RebalancePartitioner<>());
+ final int producerParallelism = environment.getTaskInfo().getNumberOfParallelSubtasks();
+ // The number of subpartitions isn't the consumer parallelism on a POINTWISE edge, so the
+ // actual consumer parallelism is passed through the deployment descriptor.
+ final int consumerParallelism = environment.getWriterConsumerParallelism(outputIndex);
Review Comment:
Thanks for the review, yes, it's intended.
Get more from another backport PR https://github.com/apache/flink/pull/29548#discussion_r4372171101
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #29549:
URL: https://github.com/apache/flink/pull/29549#discussion_r4360640087
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -1782,13 +1785,64 @@ List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createRecordWriters
private static void replaceForwardPartitionerIfConsumerParallelismDoesNotMatch(
Environment environment, NonChainedOutput streamOutput, int outputIndex) {
- if (streamOutput.getPartitioner() instanceof ForwardPartitioner
- && environment.getWriter(outputIndex).getNumberOfSubpartitions()
- != environment.getTaskInfo().getNumberOfParallelSubtasks()) {
- LOG.debug(
- "Replacing forward partitioner with rebalance for {}",
- environment.getTaskInfo().getTaskNameWithSubtasks());
- streamOutput.setPartitioner(new RebalancePartitioner<>());
+ final int producerParallelism = environment.getTaskInfo().getNumberOfParallelSubtasks();
+ // The number of subpartitions isn't the consumer parallelism on a POINTWISE edge, so the
+ // actual consumer parallelism is passed through the deployment descriptor.
+ final int consumerParallelism = environment.getWriterConsumerParallelism(outputIndex);
Review Comment:
I think this turns a forward edge overridden from 2 to 4 into a real rebalance by default; before, only consumers 0 and 2 got data. Is that intended for a patch release?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29549:
URL: https://github.com/apache/flink/pull/29549#issuecomment-6317909321
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9a35a33c2002cdff9ef7924dc6aeedac70f1a74d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9a35a33c2002cdff9ef7924dc6aeedac70f1a74d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9a35a33c2002cdff9ef7924dc6aeedac70f1a74d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "1996fanrui (via GitHub) via issues" <is...@flink.apache.org>.
1996fanrui merged PR #29549:
URL: https://github.com/apache/flink/pull/29549
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [BP-2.2][FLINK-40904][FLINK-40780][runtime] Detect FORWARD parallelism mismatch correctly and make the downgrade configurable [flink]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #29549:
URL: https://github.com/apache/flink/pull/29549#issuecomment-6403266878
Hey @RocMarshal , this backport also includes your FLINK-33123 (and its follow-up FLINK-39912), since the fix relies on it for the AdaptiveScheduler. Does it make sense to backport it to 2.2 and 1.20 (#29550)? On 1.20 it needed some adaptation.
cc @Efrat19, thanks for spotting the missing FLINK-33123.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org