[Feature][Connector-V2] Support extract partition from SeaTunnelRow fields #3085

TaoZex · 2022-10-13T03:07:11Z

Purpose of this pull request

issue：#2787

The key is specified by field name.Determine the partition of the kafka send message based on the key.

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

TaoZex · 2022-10-13T12:44:57Z

@hailin0 PTAL

EricJoy2048 · 2022-10-13T13:15:18Z

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

+            List<String> fields = Arrays.asList(seaTunnelRowType.getFieldNames());
+            String key;
+            if (fields.contains(keyField)) {
+                key = element.getField(fields.indexOf(keyField)).toString();


Don't we need to check null ？

hailin0 · 2022-10-13T13:59:30Z

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

+
+    @Override
+    public ProducerRecord<byte[], byte[]> serializeRowByKey(String key, SeaTunnelRow row) {
+        return new ProducerRecord<>(topic, key.getBytes(), jsonSerializationSchema.serialize(row));


check key is null?

check key is null?

I think we don not need check key is null, because if the key is null, kafka will send it to a random partition(random select a new partition per topic.metadata.refresh.ms).

The third property "partitioner.class" defines what class to use to determine which Partition in the Topic the message is to be sent to. This is optional, but for any non-trivial implementation you are going to want to implement a partitioning scheme. More about the implementation of this class later. If you include a value for the key but haven't defined a partitioner.class Kafka will use the default partitioner. If the key is null, then the Producer will assign the message to a random Partition.

I am sorry, you need check key is null. Because you use key.toBytes() . You can update to
return new ProducerRecord<>(topic, key == null ? null : key.getBytes(), jsonSerializationSchema.serialize(row));

hailin0 · 2022-10-13T14:03:33Z

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

@@ -21,6 +21,7 @@ By default, we will use 2pc to guarantee the message is sent to kafka exactly on
 | bootstrap.servers  | string                 | yes      | -             |
 | kafka.*            | kafka producer config  | no       | -             |
 | semantic           | string                 | no       | NON           |
+| key                | string                 | no       | -           |


Suggested change

| key | string | no | - |

| partition_key | string | no |- |

hailin0 · 2022-10-13T14:04:58Z

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

@@ -50,6 +51,21 @@ In AT_LEAST_ONCE, producer will wait for all outstanding messages in the Kafka b

 NON does not provide any guarantees: messages may be lost in case of issues on the Kafka broker and messages may be duplicated.

+### key [string]
+
+Determine the partition of the kafka send message based on the key.


Add support config filed name

EricJoy2048 · 2022-10-14T09:11:24Z

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

@@ -50,6 +51,23 @@ In AT_LEAST_ONCE, producer will wait for all outstanding messages in the Kafka b

 NON does not provide any guarantees: messages may be lost in case of issues on the Kafka broker and messages may be duplicated.

+### partition_key [string]
+
+Determine the partition of the kafka send message based on the key.


Replace with Configure which field is used as the key of the kafka message better.

EricJoy2048 · 2022-10-14T09:12:44Z

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

 - Add kafka sink doc 
 - New feature : Kafka specified partition to send 
- - New feature : Determine the partition that kafka send based on the message content
+ - New feature : Determine the partition that kafka send messag based on the message content
+ - New feature : Determine the partition of the kafka send message based on the field name


Replace with Configure which field is used as the key of the kafka message better ?

EricJoy2048 · 2022-10-14T09:16:32Z

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

+
+    @Override
+    public ProducerRecord<byte[], byte[]> serializeRowByKey(String key, SeaTunnelRow row) {
+        return new ProducerRecord<>(topic, key.getBytes(), jsonSerializationSchema.serialize(row));


I am sorry, you need check key is null. Because you use key.toBytes() . You can update to
return new ProducerRecord<>(topic, key == null ? null : key.getBytes(), jsonSerializationSchema.serialize(row));

EricJoy2048 · 2022-10-14T09:23:52Z

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

+                Object field = element.getField(fields.indexOf(keyField));
+                //If the field is null, send the message to the same partition
+                if (field == null) {
+                    key = "null";


It is not a good way. because there may be a lot of null values. If you case null to "null", it means all of null values will be write to a same partition.

You can keep null and update the code in seaTunnelRowSerializer.serializeRowByKey(key, element) like this

return new ProducerRecord<>(topic, key == null ? null : key.getBytes(), jsonSerializationSchema.serialize(row));

if the key is null, kafka will send it to a random partition(random select a new partition per topic.metadata.refresh.ms).

It' s a good way, thanks for your advice.

… into extract

hailin0 · 2022-10-15T15:49:29Z

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

@@ -74,6 +92,7 @@ public KafkaSinkWriter(
            List<KafkaSinkState> kafkaStates) {
        this.context = context;
        this.pluginConfig = pluginConfig;
+        this.seaTunnelRowType = seaTunnelRowType;


Suggested change

this.seaTunnelRowType = seaTunnelRowType;

this.partitionExtractor = createPartitionExtractor(pluginConfig, seaTunnelRowType);

Add this method

private Function<SeaTunnelRow, String> createPartitionExtractor(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) { if (!pluginConfig.hasPath(PARTITION_KEY)){ return row -> null; } String partitionKey = pluginConfig.getString(PARTITION_KEY); List<String> fieldNames = Arrays.asList(seaTunnelRowType.getFieldNames()); if (!fieldNames.contains(partitionKey)) { return row -> partitionKey; } int partitionFieldIndex = seaTunnelRowType.indexOf(partitionKey); return row -> { Object partitionFieldValue = row.getField(partitionFieldIndex); if (partitionFieldValue != null) { return partitionFieldValue.toString(); } return null; }; }

hailin0 · 2022-10-15T15:55:01Z

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

+            String keyField = pluginConfig.getString(PARTITION_KEY);
+            List<String> fields = Arrays.asList(seaTunnelRowType.getFieldNames());
+            String key;
+            if (fields.contains(keyField)) {
+                key = element.getField(fields.indexOf(keyField)).toString();
+            } else {
+                key = keyField;
+            }


Suggested change

String keyField = pluginConfig.getString(PARTITION_KEY);

List<String> fields = Arrays.asList(seaTunnelRowType.getFieldNames());

String key;

if (fields.contains(keyField)) {

key = element.getField(fields.indexOf(keyField)).toString();

} else {

key = keyField;

}

String key = partitionExtractor.apply(element);

Hi, @TaoZex I think @hailin0 gave you a good suggestion.

Thanks to @EricJoy2048 and @hailin0 the good suggestion, I will fix it.

hailin0

LGTM

hailin0 · 2022-10-17T12:17:35Z

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

@@ -21,6 +21,7 @@ By default, we will use 2pc to guarantee the message is sent to kafka exactly on
 | bootstrap.servers  | string                 | yes      | -             |
 | kafka.*            | kafka producer config  | no       | -             |
 | semantic           | string                 | no       | NON           |
+| partition_key      | string                 | no       | -           |


EricJoy2048

+1

dijiekstra · 2022-10-18T01:59:20Z

LGTM

ashulin

+1

TaoZex and others added 2 commits October 13, 2022 11:03

extract partition by field name

83a310b

Merge branch 'apache:dev' into extract

5a23186

EricJoy2048 reviewed Oct 13, 2022

View reviewed changes

hailin0 reviewed Oct 13, 2022

View reviewed changes

TaoZex and others added 2 commits October 14, 2022 16:58

Merge branch 'apache:dev' into extract

5147326

update

be2b8c4

EricJoy2048 reviewed Oct 14, 2022

View reviewed changes

TaoZex and others added 4 commits October 14, 2022 17:40

update

92c2a60

Merge branch 'apache:dev' into extract

4a19955

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

84de180

… into extract

Merge branch 'apache:dev' into extract

eceb305

TaoZex requested a review from EricJoy2048 October 15, 2022 11:29

hailin0 reviewed Oct 15, 2022

View reviewed changes

TaoZex and others added 2 commits October 17, 2022 14:38

Merge branch 'apache:dev' into extract

81a9b3c

update

3f035df

hailin0 previously approved these changes Oct 17, 2022

View reviewed changes

TaoZex and others added 2 commits October 17, 2022 20:51

Merge branch 'apache:dev' into extract

d6cfd11

update

652af1b

TaoZex dismissed hailin0’s stale review via 652af1b October 17, 2022 12:56

hailin0 approved these changes Oct 17, 2022

View reviewed changes

EricJoy2048 approved these changes Oct 18, 2022

View reviewed changes

ashulin approved these changes Oct 18, 2022

View reviewed changes

nielifeng approved these changes Oct 18, 2022

View reviewed changes

ashulin merged commit 385e1f4 into apache:dev Oct 18, 2022

CalvinKirs approved these changes Oct 18, 2022

View reviewed changes

TaoZex mentioned this pull request Oct 19, 2022

Support extract partition from SeaTunnelRow fields #2787

Closed

hailin0 mentioned this pull request Nov 4, 2022

[Improve][Connector-V2][Kafka] Support to specify multiple partition keys #3230

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][Connector-V2] Support extract partition from SeaTunnelRow fields #3085

[Feature][Connector-V2] Support extract partition from SeaTunnelRow fields #3085

TaoZex commented Oct 13, 2022

TaoZex commented Oct 13, 2022

EricJoy2048 Oct 13, 2022

hailin0 Oct 13, 2022

EricJoy2048 Oct 14, 2022

EricJoy2048 Oct 14, 2022

hailin0 Oct 13, 2022

hailin0 Oct 13, 2022

EricJoy2048 Oct 14, 2022

EricJoy2048 Oct 14, 2022

EricJoy2048 Oct 14, 2022

EricJoy2048 Oct 14, 2022

TaoZex Oct 14, 2022

hailin0 Oct 15, 2022

hailin0 Oct 15, 2022

EricJoy2048 Oct 16, 2022

TaoZex Oct 17, 2022

hailin0 left a comment

hailin0 Oct 17, 2022

EricJoy2048 left a comment

dijiekstra commented Oct 18, 2022

ashulin left a comment

	this.seaTunnelRowType = seaTunnelRowType;
	this.partitionExtractor = createPartitionExtractor(pluginConfig, seaTunnelRowType);

	\| partition_key \| string \| no \| - \|
	\| partition_key \| string \| no \| - \|

[Feature][Connector-V2] Support extract partition from SeaTunnelRow fields #3085

[Feature][Connector-V2] Support extract partition from SeaTunnelRow fields #3085

Conversation

TaoZex commented Oct 13, 2022

Purpose of this pull request

Check list

TaoZex commented Oct 13, 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

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

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

EricJoy2048 left a comment

Choose a reason for hiding this comment

dijiekstra commented Oct 18, 2022

ashulin left a comment

Choose a reason for hiding this comment