gyfora commented on code in PR #29011:
URL: https://github.com/apache/flink/pull/29011#discussion_r4002047873
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointEnvironment.java:
##########
@@ -214,11 +214,14 @@ public JobInfo getJobInfo() {
@Override
public TaskInfo getTaskInfo() {
+ // The read job's parallelism is unrelated to the operator's
maxParallelism and may exceed
+ // it (e.g. windowAll() has maxParallelism == 1); cap it to satisfy
TaskInfoImpl's
+ // maxParallelism >= numberOfParallelSubtasks invariant.
return new TaskInfoImpl(
ctx.getTaskInfo().getTaskName(),
maxParallelism,
indexOfSubtask,
- ctx.getTaskInfo().getNumberOfParallelSubtasks(),
+ Math.min(ctx.getTaskInfo().getNumberOfParallelSubtasks(),
maxParallelism),
ctx.getTaskInfo().getAttemptNumber());
Review Comment:
This issue was discovered during testing and so the fix was included here
directly
--
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]
