[Feature][Kafka] Support native format read/write kafka record by CosmosNi · Pull Request #8724 · apache/seatunnel

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

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

                && !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

+            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

+     * @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

+
+    @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

+    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

            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

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


move to static final field

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

-                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

…re-kafka-native

hailin0 · 2025-02-27T02:32:06Z

                .column(
                        PhysicalColumn.of(
                                KEY, PrimitiveByteArrayType.INSTANCE, 0, false, null, null))
-                .column(PhysicalColumn.of(OFFSET, BasicType.LONG_TYPE, 0, false, null, null))


Why delete it?

Native writing seem does not require this value.If need to keep it?

It is useful for read

hailin0 · 2025-02-27T12:31:08Z

+                                KEY, PrimitiveByteArrayType.INSTANCE, 0, false, null, null))
+                .column(PhysicalColumn.of(OFFSET, BasicType.LONG_TYPE, 0, false, null, null))
+                .column(PhysicalColumn.of(PARTITION, BasicType.INT_TYPE, 0, false, null, null))
+                .column(PhysicalColumn.of(TIMESTAMP, BasicType.LONG_TYPE, 0, false, null, null))


Adding TIMESTAMP_TYPE is nice for read

hailin0

LGTM, thanks @CosmosNi

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

63f1966

github-actions Bot added connectors-v2 e2e format 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

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

[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

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

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

a568f7b

CosmosNi requested a review from hailin0 February 26, 2025 01:53

hailin0 reviewed Feb 27, 2025

View reviewed changes

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

d88c6d9

hailin0 reviewed Feb 27, 2025

View reviewed changes

CosmosNi added 2 commits February 27, 2025 20:40

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

55c2ef7

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

729ef6e

hailin0 approved these changes Feb 27, 2025

View reviewed changes

github-actions Bot added approved reviewed labels Feb 27, 2025

Hisoka-X approved these changes Mar 3, 2025

View reviewed changes

Hisoka-X merged commit 86e2d6f into apache:dev Mar 3, 2025

Uh oh!

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

Uh oh!

litiliu Feb 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hailin0 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

litiliu Feb 14, 2025 •

edited

Loading