[FLINK-19942][Connectors / JDBC]Support sink parallelism configuration to JDBC connector #13907

zhuxiaoshang · 2020-11-03T12:37:07Z

What is the purpose of the change

Support sink parallelism configuration to JDBC connector

Brief change log

Support sink parallelism configuration to JDBC connector

Verifying this change

This change added tests and can be verified as follows:

JdbcDynamicTableFactoryTest#testJDBCSinkWithParallelism

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (no)

flinkbot · 2020-11-03T12:40:54Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit f6a4af3 (Tue Nov 03 12:40:53 UTC 2020)

Warnings:

This pull request references an unassigned Jira ticket. According to the code contribution guide, tickets need to be assigned before starting with the implementation work.

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-11-03T16:20:06Z

CI report:

f4f1cf1 UNKNOWN
9341091 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

zhuxiaoshang · 2020-11-17T09:43:15Z

cc @JingsongLi @wuchong

wuchong · 2020-12-09T05:39:30Z

cc @fsk119

zhisheng17 · 2020-12-04T06:37:10Z

...flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcExecutionOptions.java

+	public Integer getParallelism() {
+		return parallelism;
+	}
+
 	@Override
 	public boolean equals(Object o) {
 		if (this == o) {


does the equals and hash function need to be modify? @zhuxiaoshang

Yes you are right @zhisheng17 ,I ignored.

V1ncentzzZ · 2020-12-09T06:13:37Z

...flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcExecutionOptions.java

@@ -66,12 +76,13 @@ public boolean equals(Object o) {
 		JdbcExecutionOptions that = (JdbcExecutionOptions) o;
 		return batchIntervalMs == that.batchIntervalMs &&
 			batchSize == that.batchSize &&
-			maxRetries == that.maxRetries;
+			maxRetries == that.maxRetries &&
+			parallelism == that.parallelism;


Will return false when parallelism > 128?

Maybe the framework will check whether the parallelism > max parallelism limit.
Moreover where do you get the 128?IIRC,128 is the limit of max keyGroup.

My suggestion is that use equals() instead of == beacause of the == will return false when parallelism's value more than 128.

ok,I have changed it.

zhisheng17 · 2020-12-09T06:18:25Z

...flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcExecutionOptions.java

@@ -66,12 +76,13 @@ public boolean equals(Object o) {
 		JdbcExecutionOptions that = (JdbcExecutionOptions) o;
 		return batchIntervalMs == that.batchIntervalMs &&
 			batchSize == that.batchSize &&
-			maxRetries == that.maxRetries;
+			maxRetries == that.maxRetries &&
+			parallelism == that.parallelism;


@zhuxiaoshang maybe, idea will recommend to use equals() instead of ==

fsk119

Thanks for your contribution. Please rebase your branch to master.

fsk119 · 2020-12-09T08:06:09Z

...k-table-common/src/main/java/org/apache/flink/table/connector/sink/OutputFormatProvider.java

@@ -36,6 +38,23 @@ static OutputFormatProvider of(OutputFormat<RowData> outputFormat) {
 		return () -> outputFormat;


It's seem this method is useless now? If no one uses it, please delete it. Otherwise, we can just treat this method as
of(outputFormat, null)

fsk119 · 2020-12-10T07:22:15Z

@flinkbot run azure

fsk119

LGTM. The failed tests are irrelevant to this pr.

wuchong

I also left some comments.

wuchong · 2020-12-11T07:53:48Z

...k-table-common/src/main/java/org/apache/flink/table/connector/sink/OutputFormatProvider.java

 	 */
-	static OutputFormatProvider of(OutputFormat<RowData> outputFormat) {
-		return () -> outputFormat;
+	static OutputFormatProvider of(OutputFormat<RowData> outputFormat, Integer sinkParallelism) {


This is a public API. Please add a new method instead of changing the existing one, otherwise, it's not API compatible.

According to @fsk119 's suggestion,I removed the old method because it is never used.
If not i can get it back.

This is a public API. We don't know whether it is used by external projects.

ok，got it

wuchong · 2020-12-11T08:04:16Z

...flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcExecutionOptions.java

+			long batchIntervalMs,
+			int batchSize,
+			int maxRetries,
+			Integer parallelism) {


I'm not sure about adding this to JdbcExecutionOptions, because JdbcExecutionOptions is an public API for JdbcSink. For JdbcSink#sink, this may looks weird because the configured parallelism doesn't work as the returned object is SinkFunction.

Another option is adding the parallelism to JdbcDmlOption or JdbcOption, they are both internal class.

wuchong

LGTM.

Will merge this once Azure turns green.

…onnector This closes apache#13907

wuchong · 2020-12-22T02:58:17Z

The failed test is not related to this PR, it has been tracked by FLINK-20433. Will merge this issue.

…onnector This closes apache#13907 Co-authored-by: zhuxiaoshang <zhushang@qutoutiao.net>

rmetzger added review=description? component=Connectors/JDBC labels Nov 3, 2020

zhisheng17 reviewed Dec 9, 2020

View reviewed changes

V1ncentzzZ reviewed Dec 9, 2020

View reviewed changes

zhisheng17 reviewed Dec 9, 2020

View reviewed changes

fsk119 requested changes Dec 9, 2020

View reviewed changes

zhuxiaoshang force-pushed the FLINK-19942 branch from a20a546 to 5ba8c44 Compare December 9, 2020 09:35

fsk119 approved these changes Dec 11, 2020

View reviewed changes

wuchong reviewed Dec 11, 2020

View reviewed changes

zhuxiaoshang requested a review from wuchong December 11, 2020 09:43

wuchong approved these changes Dec 11, 2020

View reviewed changes

zhuxiaoshang requested a review from wuchong December 16, 2020 03:57

[FLINK-19942][jdbc] Support sink parallelism configuration for JDBC c…

9341091

…onnector This closes apache#13907

wuchong force-pushed the FLINK-19942 branch from 6acdd71 to 9341091 Compare December 21, 2020 03:07

wuchong merged commit 2ac4735 into apache:master Dec 22, 2020

meijies pushed a commit to meijies/flink that referenced this pull request Dec 28, 2020

[FLINK-19942][jdbc] Support sink parallelism configuration for JDBC c…

98ec84d

…onnector This closes apache#13907 Co-authored-by: zhuxiaoshang <zhushang@qutoutiao.net>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-19942][Connectors / JDBC]Support sink parallelism configuration to JDBC connector #13907

[FLINK-19942][Connectors / JDBC]Support sink parallelism configuration to JDBC connector #13907

zhuxiaoshang commented Nov 3, 2020

flinkbot commented Nov 3, 2020

flinkbot commented Nov 3, 2020 •

edited

zhuxiaoshang commented Nov 17, 2020

wuchong commented Dec 9, 2020

zhisheng17 Dec 4, 2020

zhuxiaoshang Dec 9, 2020

V1ncentzzZ Dec 9, 2020

zhuxiaoshang Dec 9, 2020

V1ncentzzZ Dec 9, 2020 •

edited

zhuxiaoshang Dec 9, 2020

zhisheng17 Dec 9, 2020

fsk119 left a comment

fsk119 Dec 9, 2020

fsk119 commented Dec 10, 2020

fsk119 left a comment

wuchong left a comment

wuchong Dec 11, 2020

zhuxiaoshang Dec 11, 2020

wuchong Dec 11, 2020

zhuxiaoshang Dec 11, 2020

wuchong Dec 11, 2020

wuchong left a comment

wuchong commented Dec 22, 2020

		@@ -36,6 +38,23 @@ static OutputFormatProvider of(OutputFormat<RowData> outputFormat) {
		return () -> outputFormat;

[FLINK-19942][Connectors / JDBC]Support sink parallelism configuration to JDBC connector #13907

[FLINK-19942][Connectors / JDBC]Support sink parallelism configuration to JDBC connector #13907

Conversation

zhuxiaoshang commented Nov 3, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Nov 3, 2020

Automated Checks

Review Progress

flinkbot commented Nov 3, 2020 • edited

CI report:

zhuxiaoshang commented Nov 17, 2020

wuchong commented Dec 9, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

V1ncentzzZ Dec 9, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fsk119 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fsk119 commented Dec 10, 2020

fsk119 left a comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

wuchong commented Dec 22, 2020

flinkbot commented Nov 3, 2020 •

edited

V1ncentzzZ Dec 9, 2020 •

edited