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

harveyyue · 2022-10-29T14:12:00Z

Purpose of this pull request

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

EricJoy2048 · 2022-10-29T14:27:31Z

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

@@ -51,7 +51,7 @@ 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]
+### partition_key [array]

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



Please add changed logs reference https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/connector-v2/source/HdfsFile.md

EricJoy2048 · 2022-10-29T14:34:19Z

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

+            }
+            SeaTunnelRowType keySeaTunnelRowType = new SeaTunnelRowType(keyFieldNames, keyFieldTypes);
+            return new DefaultSeaTunnelRowSerializer(this.topic, keySeaTunnelRowType, seaTunnelRowType);
+        } else {


Suggest

if () { return xx; } else if () { return xx; } else { return xx; }

replace to

if () { return xx; } if () { return xx; } return xx;

EricJoy2048 · 2022-10-29T14:35:55Z

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

+                    .filter(f -> {
+                        if (Arrays.asList(seaTunnelRowType.getFieldNames()).contains(f)) {
+                            return true;
+                        } else {


else is redundant.

harveyyue · 2022-10-30T00:53:06Z

@EricJoy2048 Done.

EricJoy2048

LGTM @hailin0 PTAL

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

EricJoy2048

LGTM

ic4y · 2022-11-01T10:09:06Z

...-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

+    @TestTemplate
+    public void testSinkKafka(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult = container.executeJob("/kafkasink_fake_to_kafka.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());


Thank you for your contribution. In addition to checking that the job exits properly, you also need to check that data is successfully written to kafka

Add the consumer data size and key message schema check

ic4y · 2022-11-01T10:13:55Z

...-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

+                .build();
+        generateTestData(row -> new ProducerRecord<>("test_topic_text", null, serializer.serialize(row)));
+        Container.ExecResult execResult = container.executeJob("/kafkasource_text_to_console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());


Ditto, it is suggest to use AssertSink or fileSink to check whether the output data is as expected

Already use AssertSink

ic4y · 2022-11-01T10:14:15Z

...-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

+        DefaultSeaTunnelRowSerializer serializer = new DefaultSeaTunnelRowSerializer("test_topic_json", SEATUNNEL_ROW_TYPE);
+        generateTestData(row -> serializer.serializeRow(row));
+        Container.ExecResult execResult = container.executeJob("/kafkasource_json_to_console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());


Already use AssertSink

hailin0 · 2022-11-04T04:00:05Z

@TaoZex please help to review

hailin0 · 2022-11-04T04:02:15Z

...atunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafkasink_fake_to_kafka.conf

+  Kafka {
+    bootstrap.servers = "kafkaCluster:9092"
+    topic = "test_topic"
+    partition_key = ["c_map","c_string"]


How to use a fixed string partition key value? eg: partition_key = 'aaa'

It is recommended to implement in this way, which is simpler for users to use

partition_key = "${c_map},${c_string}"

The value of partition_key can be a normal string or an expression to extract the field content. We have already supported single field extraction before

#3085

@hailin0 You want to keep the logic of following code snippet, right?
if (!fieldNames.contains(partitionKey)) { return row -> partitionKey; }

BTW, the fixed string partition key value will cause to write all of data to a partition of topic, specify or omit partition key(s) is good practice.

I agree

I'm worried about breaking the released feature

Perhaps rename config key to partition_key_fields is more appropriate

and need to update the examples & description in the docs

hailin0 · 2022-11-04T11:54:28Z

...unnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/log4j.properties

@@ -0,0 +1,22 @@
+#


Remove this file.

This is the common configuration of e2e
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-e2e-common/src/test/resources/log4j2.properties

hailin0 · 2022-11-04T11:55:52Z

...-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java

+        kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"))
+                .withNetwork(NETWORK)
+                .withNetworkAliases(KAFKA_HOST)
+                .withLogConsumer(new Slf4jLogConsumer(log));


Suggested change

.withLogConsumer(new Slf4jLogConsumer(log));

.withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger("confluentinc/cp-kafka:6.2.1")));

hailin0 · 2022-11-04T12:23:42Z

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

+    private List<String> createPartitionKeys(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) {
+        if (pluginConfig.hasPath(PARTITION_KEY_FIELDS)) {
+            return pluginConfig.getStringList(PARTITION_KEY_FIELDS).stream()
+                    .filter(f -> Arrays.asList(seaTunnelRowType.getFieldNames()).contains(f))
+                    .collect(Collectors.toList());
+        }
+        return null;
+    }


suggestion

private List<String> getPartitionKeyFields(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) { if (pluginConfig.hasPath(PARTITION_KEY_FIELDS)) { List<String> partitionKeyFields = pluginConfig.getStringList(PARTITION_KEY_FIELDS); List<String> rowTypeFieldNames = Arrays.asList(seaTunnelRowType.getFieldNames()); for (String partitionKeyField : partitionKeyFields) { if (!rowTypeFieldNames.contains(partitionKeyField)) { throw new IllegalArgumentException(String.format( "Partition key field not found: %s, rowType: %s", partitionKeyField, rowTypeFieldNames)); } } return partitionKeyFields; } return Collections.emptyList(); }

hailin0 · 2022-11-04T13:46:24Z

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

-    private Function<SeaTunnelRow, String> createPartitionExtractor(Config pluginConfig,
-                                                                    SeaTunnelRowType seaTunnelRowType) {
-        if (!pluginConfig.hasPath(PARTITION_KEY)){
+    private Function<SeaTunnelRow, SeaTunnelRow> createPartitionExtractor(SeaTunnelRowType seaTunnelRowType) {


suggestion

remove this method

hailin0 · 2022-11-04T13:49:47Z

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

@@ -162,12 +168,25 @@ private Properties getKafkaProperties(Config pluginConfig) {

    // todo: parse the target field from config
    private SeaTunnelRowSerializer<byte[], byte[]> getSerializer(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) {


suggestion

private SeaTunnelRowSerializer<byte[], byte[]> getSerializer(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) { if (pluginConfig.hasPath(PARTITION)){ return new DefaultSeaTunnelRowSerializer(pluginConfig.getString(TOPIC), pluginConfig.getInt(PARTITION), seaTunnelRowType); } else { return new DefaultSeaTunnelRowSerializer(pluginConfig.getString(TOPIC), getPartitionKeyFields(pluginConfig, seaTunnelRowType), seaTunnelRowType); } }

hailin0 · 2022-11-04T13:50:50Z

...n/java/org/apache/seatunnel/connectors/seatunnel/kafka/serialize/SeaTunnelRowSerializer.java

     * @param row seatunnel row
     * @return kafka record.
     */
-    ProducerRecord<K, V> serializeRowByKey(String key, SeaTunnelRow row);
+    ProducerRecord<K, V> serializeRowByKey(SeaTunnelRow key, SeaTunnelRow row);


suggestion

remove this method

Suggested change

ProducerRecord<K, V> serializeRowByKey(SeaTunnelRow key, SeaTunnelRow row);

hailin0 · 2022-11-04T13:52:04Z

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

@@ -67,13 +72,12 @@ public class KafkaSinkWriter implements SinkWriter<SeaTunnelRow, KafkaCommitInfo
    // check config
    @Override
    public void write(SeaTunnelRow element) {


suggestion

public void write(SeaTunnelRow element) { ProducerRecord<byte[], byte[]> producerRecord = seaTunnelRowSerializer.serializeRow(element); kafkaProducerSender.send(producerRecord); }

hailin0 · 2022-11-04T13:54:10Z

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

@@ -25,34 +25,42 @@

 public class DefaultSeaTunnelRowSerializer implements SeaTunnelRowSerializer<byte[], byte[]> {


suggestion

public class DefaultSeaTunnelRowSerializer implements SeaTunnelRowSerializer<byte[], byte[]> { private Integer partition; private final String topic; private final SerializationSchema keySerialization; private final SerializationSchema valueSerialization; public DefaultSeaTunnelRowSerializer(String topic, SeaTunnelRowType seaTunnelRowType) { this(topic, element -> null, createSerializationSchema(seaTunnelRowType)); } public DefaultSeaTunnelRowSerializer(String topic, Integer partition, SeaTunnelRowType seaTunnelRowType) { this(topic, seaTunnelRowType); this.partition = partition; } public DefaultSeaTunnelRowSerializer(String topic, List<String> keyFieldNames, SeaTunnelRowType seaTunnelRowType) { this(topic, createKeySerializationSchema(keyFieldNames, seaTunnelRowType), createSerializationSchema(seaTunnelRowType)); } public DefaultSeaTunnelRowSerializer(String topic, SerializationSchema keySerialization, SerializationSchema valueSerialization) { this.topic = topic; this.keySerialization = keySerialization; this.valueSerialization = valueSerialization; } @Override public ProducerRecord<byte[], byte[]> serializeRow(SeaTunnelRow row) { return new ProducerRecord<>(topic, partition, keySerialization.serialize(row), valueSerialization.serialize(row)); } private static SerializationSchema createSerializationSchema(SeaTunnelRowType rowType) { return new JsonSerializationSchema(rowType); } private static SerializationSchema createKeySerializationSchema(List<String> keyFieldNames, SeaTunnelRowType seaTunnelRowType) { int[] keyFieldIndexArr = new int[keyFieldNames.size()]; SeaTunnelDataType[] keyFieldDataTypeArr = new SeaTunnelDataType[keyFieldNames.size()]; for (int i = 0; i < keyFieldNames.size(); i++) { String keyFieldName = keyFieldNames.get(i); int rowFieldIndex = seaTunnelRowType.indexOf(keyFieldName); keyFieldIndexArr[i] = rowFieldIndex; keyFieldDataTypeArr[i] = seaTunnelRowType.getFieldType(rowFieldIndex); } SeaTunnelRowType keyType = new SeaTunnelRowType(keyFieldNames.toArray(new String[0]), keyFieldDataTypeArr); SerializationSchema keySerializationSchema = createSerializationSchema(keyType); Function<SeaTunnelRow, SeaTunnelRow> keyDataExtractor = row -> { Object[] keyFields = new Object[keyFieldIndexArr.length]; for (int i = 0; i < keyFieldIndexArr.length; i++) { keyFields[i] = row.getField(keyFieldIndexArr[i]); } return new SeaTunnelRow(keyFields); }; return row -> keySerializationSchema.serialize(keyDataExtractor.apply(row)); } }

harveyyue · 2022-11-04T16:35:27Z

@hailin0 Great suggestions! Actually, you have refactor the sink serialization.

hailin0

LGTM

ic4y

+1

Hisoka-X · 2022-11-09T09:31:28Z

Pleaes resolve conflict, Thanks!

EricJoy2048 · 2022-11-09T14:14:40Z

Hi, @harveyyue . Resolve the conflicts please.

harveyyue · 2022-11-11T16:23:06Z

Sorry late response, have fixed conflicts.

EricJoy2048 · 2022-11-12T03:28:35Z

Please merge the dev code after #3401 merged.

…keys

TyrantLucifer

@hailin0 @Hisoka-X @EricJoy2048 PTAL.

TaoZex

+1

ic4y

LGTM

EricJoy2048 reviewed Oct 29, 2022

View reviewed changes

harveyyue force-pushed the dev-kafka-2 branch from 1753424 to 002c227 Compare October 30, 2022 00:51

harveyyue force-pushed the dev-kafka-2 branch 2 times, most recently from 26d77a8 to 8cf9685 Compare October 30, 2022 07:01

EricJoy2048 previously approved these changes Oct 31, 2022

View reviewed changes

EricJoy2048 reviewed Oct 31, 2022

View reviewed changes

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

EricJoy2048 dismissed their stale review via 3e235f5 October 31, 2022 15:59

EricJoy2048 reviewed Oct 31, 2022

View reviewed changes

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

EricJoy2048 previously approved these changes Nov 1, 2022

View reviewed changes

ic4y reviewed Nov 1, 2022

View reviewed changes

EricJoy2048 added the Waiting for code update label Nov 2, 2022

hailin0 reviewed Nov 4, 2022

View reviewed changes

harveyyue dismissed EricJoy2048’s stale review via 22cfbbb November 4, 2022 05:54

harveyyue force-pushed the dev-kafka-2 branch 2 times, most recently from 3cb846d to 30a2305 Compare November 4, 2022 11:41

hailin0 reviewed Nov 4, 2022

View reviewed changes

harveyyue force-pushed the dev-kafka-2 branch from cc5bf74 to 053b177 Compare November 4, 2022 16:24

hailin0 previously approved these changes Nov 5, 2022

View reviewed changes

ic4y previously approved these changes Nov 7, 2022

View reviewed changes

Hisoka-X removed the Waiting for code update label Nov 9, 2022

harveyyue dismissed stale reviews from ic4y and hailin0 via c7eaae7 November 11, 2022 16:12

harveyyue force-pushed the dev-kafka-2 branch 2 times, most recently from c7eaae7 to 96ce728 Compare November 11, 2022 16:18

[Improve][Connector-V2][Kafka] Support to specify multiple partition …

7715ba5

…keys

harveyyue force-pushed the dev-kafka-2 branch from 96ce728 to 7715ba5 Compare November 17, 2022 03:25

TyrantLucifer approved these changes Nov 17, 2022

View reviewed changes

github-actions bot added approved reviewed labels Nov 17, 2022

TaoZex approved these changes Nov 17, 2022

View reviewed changes

ic4y approved these changes Nov 17, 2022

View reviewed changes

ic4y merged commit f65f44f into apache:dev Nov 17, 2022

	.withLogConsumer(new Slf4jLogConsumer(log));
	.withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger("confluentinc/cp-kafka:6.2.1")));

		@@ -162,12 +168,25 @@ private Properties getKafkaProperties(Config pluginConfig) {

		// todo: parse the target field from config
		private SeaTunnelRowSerializer<byte[], byte[]> getSerializer(Config pluginConfig, SeaTunnelRowType seaTunnelRowType) {

		@@ -25,34 +25,42 @@

		public class DefaultSeaTunnelRowSerializer implements SeaTunnelRowSerializer<byte[], byte[]> {

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

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

Conversation

harveyyue commented Oct 29, 2022

Purpose of this pull request

Check list

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

harveyyue commented Oct 30, 2022

EricJoy2048 left a comment

Choose a reason for hiding this comment

EricJoy2048 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

Choose a reason for hiding this comment

hailin0 commented Nov 4, 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

harveyyue Nov 4, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hailin0 Nov 4, 2022 • edited Loading

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

harveyyue commented Nov 4, 2022

hailin0 left a comment

Choose a reason for hiding this comment

ic4y left a comment

Choose a reason for hiding this comment

Hisoka-X commented Nov 9, 2022

EricJoy2048 commented Nov 9, 2022

harveyyue commented Nov 11, 2022

EricJoy2048 commented Nov 12, 2022

TyrantLucifer left a comment

Choose a reason for hiding this comment

TaoZex left a comment

Choose a reason for hiding this comment

ic4y left a comment

Choose a reason for hiding this comment

harveyyue Nov 4, 2022 •

edited

Loading

hailin0 Nov 4, 2022 •

edited

Loading