rkhachatryan commented on code in PR #29272:
URL: https://github.com/apache/flink/pull/29272#discussion_r4088147465
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -2046,13 +2048,52 @@
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();
+ final int consumerParallelism =
+ environment.getWriter(outputIndex).getNumberOfSubpartitions();
Review Comment:
`getNumberOfSubpartitions()` isn't the consumer parallelism on a `POINTWISE`
(forward) edge. In a static graph it's the size of the consumer vertex group
for this producer subtask
(`IntermediateResultPartition#getNumberOfSubpartitions`), so it's 1 when
producer and consumer parallelism are equal. In a dynamic graph (adaptive
batch) a forward result always has exactly 1 subpartition
(computeNumberOfSubpartitionsForDynamicGraph). So the != producerParallelism
check is wrong both ways.
False positive: any non-chained forward edge with parallelism > 1 and no
rescale counts as a mismatch. Before this PR that was harmless, since the
result was a RebalancePartitioner over a single channel. Now FAIL rejects such
jobs and KEEP_FORWARD logs a WARN for every subtask.
False negative: producer 2 → consumer 4 gives 2 subpartitions per producer,
which equals the producer parallelism of 2, so nothing is detected. FAIL
doesn't fail and REBALANCE doesn't rebalance, and half the consumers get no
data.
I checked this with a MiniCluster ITCase on this branch, with mode FAIL,
source → forward() → map (startNewChain):
| Case | Result
|
|----------------------------------------------------------------|-----------------------------------------------------------------|
| streaming 4 → 4, no rescale | fails:
producer parallelism 4 != consumer parallelism 1 |
| batch 4 → 4, no rescale | fails,
same error |
| streaming 2 → 4 (via pipeline.jobvertex-parallelism-overrides) | succeeds,
records per consumer subtask {0=500, 1=0, 2=500, 3=0} |
| streaming 1 → 2 (the case covered by StreamTaskTest) | fails, as
expected |
The existing tests only cover 1 → 2, where the two quantities happen to
coincide.
The error and WARN messages have the same mix-up: they report the
subpartition count as "consumer parallelism", e.g. "consumer parallelism 1" for
a consumer running at 4.
I think the check needs the actual consumer vertex parallelism, which I
don't believe is available in the task today, so it may have to be passed
through the deployment descriptor or NonChainedOutput. Alternatively, the
mismatch could be decided at scheduling time. Could you also add tests for
equal parallelism > 1 (streaming and batch) and for N → M with N > 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
