JingsongLi commented on a change in pull request #13789:
URL: https://github.com/apache/flink/pull/13789#discussion_r513152799
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##########
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+ val inputParallelism = inputTransformation.getParallelism
+ val taskParallelism = env.getParallelism
Review comment:
`env.getMaxParallelism`?
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##########
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+ val inputParallelism = inputTransformation.getParallelism
+ val taskParallelism = env.getParallelism
+ val parallelism = if
(runtimeProvider.isInstanceOf[ParallelismProvider])
runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue()
+ else inputParallelism
+
+ if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new
RuntimeException(s"the configured sink parallelism: $parallelism should not be
less than zero or equal to zero")
+ if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw
new RuntimeException(s"the configured sink parallelism: $parallelism is larger
than the task max parallelism: $taskParallelism")
Review comment:
We don't need verify `env.getMaxParallelism`, DataStream layer will
verify it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
