flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
/**
	 * Verifies conditions in regards to parallelism and maxParallelism that must be met when restoring state.
	 *
	 * @param operatorState      state to restore
	 * @param executionJobVertex task for which the state should be restored
	 */
Verifies conditions in regards to parallelism and maxParallelism that must be met when restoring state.
@param operatorState      state to restore
@param executionJobVertex task for which the state should be restored
Verifies conditions in regards to parallelism and maxParallelism that must be met when restoring state . @param operatorState state to restore @param executionJobVertex task for which the state should be restored
private static void checkParallelismPreconditions(OperatorState operatorState, ExecutionJobVertex executionJobVertex) {
		if (operatorState.getMaxParallelism() < executionJobVertex.getParallelism()) {
			throw new IllegalStateException("The state for task " + executionJobVertex.getJobVertexId() +
				" can not 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 (operatorState.getMaxParallelism() != executionJobVertex.getMaxParallelism()) {
			if (!executionJobVertex.isMaxParallelismConfigured()) {
				LOG.debug("Overriding maximum parallelism for JobVertex {} from {} to {}",
					executionJobVertex.getJobVertexId(), executionJobVertex.getMaxParallelism(), operatorState.getMaxParallelism());
				executionJobVertex.setMaxParallelism(operatorState.getMaxParallelism());
			} else {
				throw new IllegalStateException("The maximum parallelism (" +
					operatorState.getMaxParallelism() + ") with which the latest " +
					"checkpoint of the execution job vertex " + executionJobVertex +
					" has been taken and the current maximum parallelism (" +
					executionJobVertex.getMaxParallelism() + ") changed. This " +
					"is currently not supported.");
			}
		}
	}
private static void checkParallelismPreconditions ( OperatorState operatorState , ExecutionJobVertex executionJobVertex ) { if ( operatorState . getMaxParallelism ( ) < executionJobVertex . getParallelism ( ) ) { throw new IllegalStateException ( "The state for task " + executionJobVertex . getJobVertexId ( ) + " can not 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 ( operatorState . getMaxParallelism ( ) != executionJobVertex . getMaxParallelism ( ) ) { if ( ! executionJobVertex . isMaxParallelismConfigured ( ) ) { LOG . debug ( "Overriding maximum parallelism for JobVertex {} from {} to {}" , executionJobVertex . getJobVertexId ( ) , executionJobVertex . getMaxParallelism ( ) , operatorState . getMaxParallelism ( ) ) ; executionJobVertex . setMaxParallelism ( operatorState . getMaxParallelism ( ) ) ; } else { throw new IllegalStateException ( "The maximum parallelism (" + operatorState . getMaxParallelism ( ) + ") with which the latest " + "checkpoint of the execution job vertex " + executionJobVertex + " has been taken and the current maximum parallelism (" + executionJobVertex . getMaxParallelism ( ) + ") changed. This " + "is currently not supported." ) ; } } }
Verifies conditions in regards to parallelism and maxParallelism that must be met when restoring state.
Verifies conditions in regards to parallelism and maxParallelism that must be met when restoring state .
//----------------------------------------max parallelism preconditions------------------------------------- // check that the number of key groups have not changed or if we need to override it to satisfy the restored state // if the max parallelism was not explicitly specified by the user, we derive it from the state // if the max parallelism was explicitly specified, we complain on mismatch