[FLINK-35713][cdc-compose] Add sink PARALLELISM for flink-cdc. #3438
Conversation
|
Thanks for @proletarians' contribution! Kindly remind that |
There was a problem hiding this comment.
In the future, defining multiple sink in one single pipeline job might be supported, and one might want to define parallelisms for each sink individually. Will it be better if we put sink.parallelism as an option in sink: block instead of a global pipeline option?
There was a problem hiding this comment.
Got it! I will follow up on this.
| @@ -93,6 +93,7 @@ private FlinkPipelineComposer(StreamExecutionEnvironment env, boolean isBlocking | |||
| @Override | |||
| public PipelineExecution compose(PipelineDef pipelineDef) { | |||
| int parallelism = pipelineDef.getConfig().get(PipelineOptions.PIPELINE_PARALLELISM); | |||
| int sinkParallelism = pipelineDef.getConfig().get(PipelineOptions.SINK_PARALLELISM); | |||
There was a problem hiding this comment.
What if users do not set the SINK_PARALLELISM setting?
| ConfigOptions.key("sink.parallelism") | ||
| .intType() | ||
| .noDefaultValue() | ||
| .withDescription("Parallelism of the sink in the pipeline"); |
There was a problem hiding this comment.
Please add some tests for the setting.
|
Hi @proletarians, you may want to create individual branches for each PR to avoid mixing different changes together. |
|
This pull request has been automatically marked as stale because it has not had recent activity for 60 days. It will be closed in 30 days if no further activity occurs. |
No description provided.