[Improve][Connector-V2][Kafka Sink]custom partition #2889

TaoZex · 2022-09-25T14:21:06Z

Purpose of this pull request

The following code supports two user-defined partition functions:
The first type: the user specifies the partition to send through the partition field in the config (for example, if partition 2 is specified, then the messages are sent to partition 2)

The second: the user customizes the partition through the assign_partitions field in the config
For example, there are 5 partitions in total, and the assign_partitions field in config is as follows:
assignpartitions = ["shoe", "clothing"]
Then the message containing "shoe" will be sent to partition 0 (because "shoe" is subscripted as 0 in assign_partitions), and the message containing "clothing" will be sent to partition 1,
For other messages, the hash algorithm will be used to divide them into the remaining partitions.

Check list

Code changed are covered with tests, or it does not need tests for reason:
If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs

hailin0 · 2022-09-26T02:59:57Z

Can you add docs?

https://github.com/apache/incubator-seatunnel/tree/dev/docs/en/connector-v2/source

TaoZex · 2022-09-26T03:03:38Z

Can you add docs?

https://github.com/apache/incubator-seatunnel/tree/dev/docs/en/connector-v2/source

Sure, I'll finish writing the documentation.Thanks for your review.

EricJoy2048 · 2022-09-26T11:23:37Z

Please solve the conflicting files.

ashulin

Please check your code style:

mvn checkstyle:check -T 1C

Resolving code conflicts:

git rebase refs/remotes/${apache remote-name}/dev

git push --force-with-lease

TaoZex · 2022-09-26T15:39:41Z

Please solve the conflicting files.

Thanks for your comment, I will fix it.

TaoZex · 2022-09-26T15:40:13Z

Please check your code style:
mvn checkstyle:check -T 1C
Resolving code conflicts:
git rebase refs/remotes/${apache remote-name}/dev

git push --force-with-lease

Thanks for your review, I will follow your instructions.

TaoZex · 2022-09-27T04:59:00Z

The CI pipeline failed, and the error location is in connector-jdbc-spark-e2e, but I did not submit the code related to jdbc. Before adding the kafka sink document, the code also passed the CI pipeline. Is it a problem with the pipeline or the code?

EricJoy2048 · 2022-09-27T05:39:38Z

The CI pipeline failed, and the error location is in connector-jdbc-spark-e2e, but I did not submit the code related to jdbc. Before adding the kafka sink document, the code also passed the CI pipeline. Is it a problem with the pipeline or the code?

It maybe cause by #2907. After #2907 has been merged, I will rerun the failed jobs.

TaoZex · 2022-09-27T05:41:04Z

The CI pipeline failed, and the error location is in connector-jdbc-spark-e2e, but I did not submit the code related to jdbc. Before adding the kafka sink document, the code also passed the CI pipeline. Is it a problem with the pipeline or the code?

It maybe cause by #2907. After #2907 has been merged, I will rerun the failed jobs.

Thank you for your reply, it answered my doubts.

TaoZex · 2022-09-27T14:12:07Z

I have fixed it, please review it again.Thanks.

hailin0 · 2022-09-28T03:58:55Z

...org/apache/seatunnel/connectors/seatunnel/kafka/serialize/DefaultSeaTunnelRowSerializer.java

+        this.topic = topic;
+        this.partation = partation;
+        this.jsonSerializationSchema = new JsonSerializationSchema(seaTunnelRowType);


Suggested change

this.topic = topic;

this.partation = partation;

this.jsonSerializationSchema = new JsonSerializationSchema(seaTunnelRowType);

this(topic, seaTunnelRowType);

this.partation = partation;

hailin0 · 2022-09-28T04:00:39Z

...ka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/CustomPartitioner.java

+            }
+        }
+        //Choose one of the remaining partitions according to the hashcode.
+        return (Math.abs(message.hashCode()) % (numPartitions - assignPartitionsSize)) + assignPartitionsSize;


Math.abs(message.hashCode()) -> message.hashCode() & Integer.MAX_VALUE

reference

#2921

hailin0 · 2022-09-28T04:06:43Z

...afka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java

+    private static List<String> ASSIGNPARTATIONS;
+
+    public static List<String> getASSIGNPARTATIONS() {
+        return ASSIGNPARTATIONS;
+    }


hailin0 · 2022-09-28T04:07:32Z

...afka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java

+            this.partition = pluginConfig.getInt(PARTITION);
+        }
+        if (pluginConfig.hasPath(ASSIGN_PARTITIONS)) {
+            ASSIGNPARTATIONS = pluginConfig.getStringList(ASSIGN_PARTITIONS);


Suggested change

ASSIGNPARTATIONS = pluginConfig.getStringList(ASSIGN_PARTITIONS);

CustomPartitioner.setAssignPartitions(pluginConfig.getStringList(ASSIGN_PARTITIONS));

hailin0 · 2022-09-28T04:09:16Z

...ector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/Config.java

+    /**
+     * Determine the partition to send based on the content of the message.
+     */
+    public static final String ASSIGN_PARTITIONS = "assign_partitions";


...afka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaSinkWriter.java

…nto par � Conflicts: � docs/en/connector-v2/source/IoTDB.md

This reverts commit 8442fd7

docs/en/connector-v2/sink/kafka.md

EricJoy2048 · 2022-09-30T09:34:13Z

docs/en/connector-v2/sink/kafka.md

+
+Kafka Brokers List.
+
+### producer [string]


Replace with ### kafka.* [kafka producer config].

EricJoy2048 · 2022-10-05T06:09:59Z

@hailin0 PTAL

hailin0

LGTM

hailin0 · 2022-10-06T08:50:38Z

...ka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/CustomPartitioner.java

+import java.util.List;
+import java.util.Map;
+
+public class CustomPartitioner implements Partitioner {


rename CustomPartitioner to MessageContentPartitioner

hailin0 · 2022-10-06T14:19:05Z

docs/en/connector-v2/sink/kafka.md

+
+Then the message containing "shoe" will be sent to partition zero ,because "shoe" is subscripted as zero in assign_partitions, and the message containing "clothing" will be sent to partition one.For other messages, the hash algorithm will be used to divide them into the remaining partitions.
+
+This function by `CustomPartitioner` class implements `org.apache.kafka.clients.producer.Partitioner` interface.If we need custom partitions, we need to implement this interface as well.


rename CustomPartitioner ?

custom partition

5b1ee04

Hisoka-X added the Waiting for code update label Sep 26, 2022

update kafka sink doc

ea581e0

EricJoy2048 changed the title ~~custom partition~~ [Improve][Connector-V2][Kafka Sink]custom partition Sep 26, 2022

update

572a9af

ashulin reviewed Sep 26, 2022

View reviewed changes

Merge branch 'dev' into par

b967635

TaoZex and others added 2 commits September 27, 2022 15:26

update connector V2 common-options

8442fd7

Update kafka.md

078aabb

hailin0 reviewed Sep 28, 2022

View reviewed changes

TaoZex and others added 9 commits September 28, 2022 15:53

Merge branch 'apache:dev' into par

8614fcc

Merge branch 'apache:dev' into par

a461055

Merge branch 'par' of https://github.com/TaoZex/incubator-seatunnel i…

de2fe6b

…nto par � Conflicts: � docs/en/connector-v2/source/IoTDB.md

update

2f09ed7

Revert "update connector V2 common-options"

4f8ec2c

This reverts commit 8442fd7

update

c0a54c8

update

0e11275

update

3dfbd08

update

32ca0b3

TaoZex requested a review from hailin0 September 29, 2022 14:12

TaoZex mentioned this pull request Sep 29, 2022

[Doc] [Connector-V2] Add connector-v2 kafka source doc #2953

Merged

3 tasks

TaoZex requested a review from ashulin September 30, 2022 03:50

EricJoy2048 reviewed Sep 30, 2022

View reviewed changes

docs/en/connector-v2/sink/kafka.md Outdated Show resolved Hide resolved

TaoZex requested review from EricJoy2048 and removed request for ashulin September 30, 2022 09:30

EricJoy2048 reviewed Sep 30, 2022

View reviewed changes

TaoZex and others added 4 commits September 30, 2022 18:31

update

9c3d016

Merge branch 'apache:dev' into par

90dc08f

update

eb50ffd

Merge remote-tracking branch 'origin/par' into par

3527313

TaoZex requested a review from EricJoy2048 October 1, 2022 02:13

TaoZex and others added 3 commits October 5, 2022 00:54

Merge branch 'apache:dev' into par

51b9229

update

189e987

update

0f92c00

TaoZex requested review from hailin0 and EricJoy2048 and removed request for EricJoy2048 and hailin0 October 4, 2022 17:17

update

9e3a082

hailin0 reviewed Oct 6, 2022

View reviewed changes

TaoZex and others added 2 commits October 6, 2022 21:10

update

dea8966

Merge branch 'apache:dev' into par

b145fa9

hailin0 reviewed Oct 6, 2022

View reviewed changes

TaoZex and others added 4 commits October 6, 2022 23:07

update

8cb6087

Merge branch 'apache:dev' into par

dee3a8c

Merge branch 'apache:dev' into par

8ddb6dc

Merge branch 'apache:dev' into par

9e83250

TaoZex closed this Oct 8, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Improve][Connector-V2][Kafka Sink]custom partition #2889

[Improve][Connector-V2][Kafka Sink]custom partition #2889

TaoZex commented Sep 25, 2022 •

edited

Loading

hailin0 commented Sep 26, 2022

TaoZex commented Sep 26, 2022

EricJoy2048 commented Sep 26, 2022

ashulin left a comment

TaoZex commented Sep 26, 2022

TaoZex commented Sep 26, 2022

TaoZex commented Sep 27, 2022

EricJoy2048 commented Sep 27, 2022

TaoZex commented Sep 27, 2022

TaoZex commented Sep 27, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

EricJoy2048 Sep 30, 2022

EricJoy2048 commented Oct 5, 2022

hailin0 left a comment

hailin0 Oct 6, 2022

hailin0 Oct 6, 2022

	ASSIGNPARTATIONS = pluginConfig.getStringList(ASSIGN_PARTITIONS);
	CustomPartitioner.setAssignPartitions(pluginConfig.getStringList(ASSIGN_PARTITIONS));


		Then the message containing "shoe" will be sent to partition zero ,because "shoe" is subscripted as zero in assign_partitions, and the message containing "clothing" will be sent to partition one.For other messages, the hash algorithm will be used to divide them into the remaining partitions.

		This function by `CustomPartitioner` class implements `org.apache.kafka.clients.producer.Partitioner` interface.If we need custom partitions, we need to implement this interface as well.

[Improve][Connector-V2][Kafka Sink]custom partition #2889

[Improve][Connector-V2][Kafka Sink]custom partition #2889

Conversation

TaoZex commented Sep 25, 2022 • edited Loading

Purpose of this pull request

Check list

hailin0 commented Sep 26, 2022

TaoZex commented Sep 26, 2022

EricJoy2048 commented Sep 26, 2022

ashulin left a comment

Choose a reason for hiding this comment

TaoZex commented Sep 26, 2022

TaoZex commented Sep 26, 2022

TaoZex commented Sep 27, 2022

EricJoy2048 commented Sep 27, 2022

TaoZex commented Sep 27, 2022

TaoZex commented Sep 27, 2022

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

Choose a reason for hiding this comment

EricJoy2048 commented Oct 5, 2022

hailin0 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TaoZex commented Sep 25, 2022 •

edited

Loading