Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/4125#discussion_r122686915
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
---
@@ -225,7 +225,16 @@ private void assignAttemptState(ExecutionJobVertex
executionJobVertex, List<Oper
public void checkParallelismPreconditions(List<OperatorState>
operatorStates, ExecutionJobVertex executionJobVertex) {
-
+ //----------------------------------------parallelism compare
preconditions-------------------------------------
+
+ // if the max parallelism is lower than parallelism, we will
throw an exception.
+ if (executionJobVertex.getMaxParallelism() <
executionJobVertex.getParallelism()) {
+ throw new IllegalArgumentException("JobVertex: " +
executionJobVertex.getJobVertex() +
--- End diff --
We don't usually refer to JobVertices in exceptions, a more user-known
"equivalent" are tasks.
We should also throw an IllegalStateException to be consistent with
existing exceptions in this area.
Finally, let's reword this a bit:
"The state for task (<executionJobVertex.getJobVertexId()>) cannot be
restored. The maximum parallelism (<operatorState.getMaxParallelism()>) of the
restored state is lower than the configured parallelism
(<executionJobVertex.getParallelism()>). Please reduce the parallelism of the
task to be lower or equal to the maximum parallelism."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
