[Feature][Kafka] Support native format read/write kafka record #8724

CosmosNi · 2025-02-14T08:28:11Z

close #8651

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

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/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update plugin-mapping.properties and add new connector information in it
2. Update the pom file of seatunnel-dist
3. Add ci label in label-scope-conf
4. Add e2e testcase in seatunnel-e2e
5. Update connector plugin_config
Update the release-note.

litiliu · 2025-02-14T09:02:30Z

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

@@ -80,6 +83,7 @@ public KafkaSinkWriter(
                && !CollectionUtils.isEmpty(pluginConfig.get(ASSIGN_PARTITIONS))) {
            MessageContentPartitioner.setAssignPartitions(pluginConfig.get(ASSIGN_PARTITIONS));
        }
+        isNative = pluginConfig.get(IS_NATIVE) != null && pluginConfig.get(IS_NATIVE);


Is it better to use Boolean.True.equals(pluginConfig.get(IS_NATIVE))?

litiliu · 2025-02-14T09:17:58Z

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

+            ConsumerRecord<String, String> oldRecord = data.get(i);
+            ConsumerRecord<String, String> newRecord = data.get(i);
+            Assertions.assertEquals(oldRecord.key(), newRecord.key());
+            Assertions.assertEquals(oldRecord.headers(), newRecord.headers());
+            Assertions.assertEquals(oldRecord.partition(), newRecord.partition());
+            Assertions.assertEquals(oldRecord.timestamp(), newRecord.timestamp());
+            Assertions.assertEquals(oldRecord.timestamp(), newRecord.timestamp());
+            Assertions.assertEquals(oldRecord.value(), newRecord.value());


Seems the oldRecord and newRecord are referring to the same object, what's the meaning of those assertions?

litiliu · 2025-02-14T09:23:22Z

.../seatunnel/format/compatible/kafka/connect/json/NativeKafkaConnectDeserializationSchema.java

+     * @param out
+     */
+    public void deserialize(ConsumerRecord<byte[], byte[]> msg, Collector<SeaTunnelRow> out) {
+        tryInitConverter();


Could we try to move the init logic to the constructor?

litiliu · 2025-02-14T09:31:36Z

.../seatunnel/format/compatible/kafka/connect/json/NativeKafkaConnectDeserializationSchema.java

+
+    @Override
+    public SeaTunnelRow deserialize(byte[] message) throws IOException {
+        throw new UnsupportedOperationException(


Why is it not suggested that this method be invoked?

hailin0 · 2025-02-14T09:39:39Z

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

+    is_native = true
+    topic = "test_topic"
+    bootstrap.servers = "kafkaCluster:9092"
+    kafka.config = {
+      client.id = client_1
+      max.poll.records = 500
+      auto.offset.reset = "earliest"
+      enable.auto.commit = "false"
+    }
+    format = "COMPATIBLE_KAFKA_CONNECT_JSON"


Suggested change

is_native = true

topic = "test_topic"

bootstrap.servers = "kafkaCluster:9092"

kafka.config = {

client.id = client_1

max.poll.records = 500

auto.offset.reset = "earliest"

enable.auto.commit = "false"

}

format = "COMPATIBLE_KAFKA_CONNECT_JSON"

topic = "test_topic"

bootstrap.servers = "kafkaCluster:9092"

kafka.config = {

client.id = client_1

max.poll.records = 500

auto.offset.reset = "earliest"

enable.auto.commit = "false"

}

format = "NATIVE"

Add NATIVE into MessageFormat

https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/MessageFormat.java#L30

…-kafka-native # Conflicts: # seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

litiliu · 2025-02-17T03:28:57Z

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

@@ -234,6 +275,7 @@ private static SerializationSchema createSerializationSchema(
            ReadonlyConfig pluginConfig) {
        switch (format) {
            case JSON:
+            case NATIVE:


if the format is native, we shouldn't expect the content to be JSON

Yes, content is not json.This is used to process producerRecord.

just like:SeatTunnelRowType<topic, partition, offset, timestamp, headers<map<string,string>>, key, value>

hailin0 · 2025-02-17T13:11:28Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

+
+        if (format == MessageFormat.NATIVE) {
+            tableSchema =
+                    TableSchema.builder()


move to static final field

hailin0 · 2025-02-17T13:16:25Z

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

-                seaTunnelRowSerializer.serializeRow(element);
+        ProducerRecord<byte[], byte[]> producerRecord;
+        if (isNative) {
+            producerRecord = seaTunnelRowSerializer.serializeNativeRow(element, seaTunnelRowType);


Why not override topicExtractor, partitionExtractor, ...

remove this if-else

# Conflicts: # seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

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

…re-kafka-native

hailin0 · 2025-02-24T03:48:20Z

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

@@ -170,18 +170,22 @@ private Properties getKafkaProperties(ReadonlyConfig pluginConfig) {
    private SeaTunnelRowSerializer<byte[], byte[]> getSerializer(
            ReadonlyConfig pluginConfig, SeaTunnelRowType seaTunnelRowType) {
        MessageFormat messageFormat = pluginConfig.get(FORMAT);
+        String topic = pluginConfig.get(TOPIC);
+        if (MessageFormat.NATIVE.equals(messageFormat)) {
+            return DefaultSeaTunnelRowSerializer.create(topic, messageFormat, seaTunnelRowType);


check input fields & types when sink init

[Feature][Kafka] Support native format read/write kafka record

63f1966

github-actions bot added connectors-v2 e2e format kafka labels Feb 14, 2025

[Feature][Kafka] Support native format read/write kafka record

56aa7a1

github-actions bot added the document label Feb 14, 2025

litiliu reviewed Feb 14, 2025

View reviewed changes

hailin0 reviewed Feb 14, 2025

View reviewed changes

CosmosNi added 6 commits February 14, 2025 22:17

[Feature][Kafka] Support native format read/write kafka record

afe9986

[Feature][Kafka] Support native format read/write kafka record

21a6d71

[Feature][Kafka] Support native format read/write kafka record

3e35347

[Feature][Kafka] Support native format read/write kafka record

6059e42

Merge remote-tracking branch 'refs/remotes/upstream/dev' into feature…

44d2da8

…-kafka-native # Conflicts: # seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

[Feature][Kafka] Support native format read/write kafka record

e03d62a

litiliu reviewed Feb 17, 2025

View reviewed changes

CosmosNi added 4 commits February 17, 2025 14:05

[Feature][Kafka] Support native format read/write kafka record

959d263

[Feature][Kafka] Support native format read/write kafka record

91dce70

[Feature][Kafka] Support native format read/write kafka record

db197b8

[Feature][Kafka] Support native format read/write kafka record

273f8c5

hailin0 reviewed Feb 17, 2025

View reviewed changes

CosmosNi added 4 commits February 18, 2025 10:46

[Feature][Kafka] Support native format read/write kafka record

585ad48

[Feature][Kafka] Support native format read/write kafka record

5b48456

Merge remote-tracking branch 'upstream/dev' into feature-kafka-native

bb2c96f

# Conflicts: # seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

[Feature][Kafka] Support native format read/write kafka record

0e92b1c

hailin0 reviewed Feb 19, 2025

View reviewed changes

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

[Feature][Kafka] Support native format read/write kafka record

e4ff9c4

Merge remote-tracking branch 'origin/feature-kafka-native' into featu…

84c904f

…re-kafka-native

CosmosNi requested a review from hailin0 February 21, 2025 07:22

hailin0 reviewed Feb 24, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][Kafka] Support native format read/write kafka record #8724

[Feature][Kafka] Support native format read/write kafka record #8724

CosmosNi commented Feb 14, 2025

litiliu Feb 14, 2025 •

edited

Loading

litiliu Feb 14, 2025

litiliu Feb 14, 2025

litiliu Feb 14, 2025

hailin0 Feb 14, 2025

hailin0 Feb 14, 2025

litiliu Feb 17, 2025

CosmosNi Feb 17, 2025

CosmosNi Feb 17, 2025

hailin0 Feb 17, 2025

hailin0 Feb 17, 2025

hailin0 Feb 24, 2025

[Feature][Kafka] Support native format read/write kafka record #8724

Are you sure you want to change the base?

[Feature][Kafka] Support native format read/write kafka record #8724

Conversation

CosmosNi commented Feb 14, 2025

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

litiliu Feb 14, 2025 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

litiliu Feb 14, 2025 •

edited

Loading