[Feature][connectors-v2][kafka] Kafka supports custom schema #2371 #2783

eyys · 2022-09-18T15:06:43Z

Purpose of this pull request

[Feature][seatunnel-connectors-v2][connector-kafka] Kafka supports custom schema #2371
The last PR that was closed #2720

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

此拉取请求的目的

Kafka 支持自定义Schema #2371
上一个关闭的PR #2720

检查列表

添加e2e测试
添加example测试

hailin0 · 2022-09-19T09:30:20Z

...nnector-v2-e2e/connector-kafka-flink-e2e/src/test/resources/kafka/kafkasouce_to_console.conf

+      fields {
+        name = "string"
+        age = "int"
+      }


test all datatype?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-iceberg-flink-e2e/src/test/resources/iceberg/iceberg_source.conf#L29

Ok, thanks for the Pointers

hailin0 · 2022-09-19T09:31:17Z

...nnector-v2-e2e/connector-kafka-flink-e2e/src/test/resources/kafka/kafkasouce_to_console.conf

+}
+
+sink {
+  Console {}


Add Assert check row data type?

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-iceberg-flink-e2e/src/test/resources/iceberg/iceberg_source.conf#L61

hailin0 · 2022-09-19T09:32:44Z

...nnector-v2-e2e/connector-kafka-flink-e2e/src/test/resources/kafka/kafkasouce_to_console.conf

+    sql {
+      sql = "select name,age from kafka"
+    }


remove or use select * from kafka selected all fields

hailin0 · 2022-09-19T09:34:00Z

...-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

+            }else{
+                format = config.getString(FORMAT);
+                this.deserializationSchema = null;
+            }
+        }else {


check codestyle

hailin0 · 2022-09-19T09:40:07Z

...nnector-v2-e2e/connector-kafka-flink-e2e/src/test/resources/kafka/kafkasouce_to_console.conf

+      }
+    }
+    topic = "test_csv"
+    bootstrap.server = "hadoop101:9092,hadoop102:9092,hadoop103:9092"


You need to start the kafka test container in the test case

Connect your test container and initialize the data

Test your logic

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/mongodb/MongodbSourceToConsoleIT.java#L70
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/mongodb_to_console.conf#L33

Okay, I'll try to set it up

hailin0 · 2022-09-19T09:40:43Z

...nector-v2-e2e/connector-kafka-spark-e2e/src/test/resources/kafka/kafkasource_to_console.conf

+    result_table_name = "kafka"
+    schema = {
+      fields {
+        name = "string"


hailin0 · 2022-09-19T09:40:50Z

...nector-v2-e2e/connector-kafka-spark-e2e/src/test/resources/kafka/kafkasource_to_console.conf

+      }
+    }
+    topic = "test_csv"
+    bootstrap.server = "hadoop101:9092,hadoop102:9092,hadoop103:9092"


hailin0 · 2022-09-19T09:40:59Z

...nector-v2-e2e/connector-kafka-spark-e2e/src/test/resources/kafka/kafkasource_to_console.conf

+
+transform {
+    sql {
+      sql = "select name,age from kafka"


hailin0 · 2022-09-19T09:41:09Z

...nector-v2-e2e/connector-kafka-spark-e2e/src/test/resources/kafka/kafkasource_to_console.conf

+
+sink {
+  Console {}
+}


hailin0 · 2022-09-24T09:26:53Z

recheck code

ashulin

Can you move e2e IT to seatunnel-connector-v2-e2e module? you can see #2924.

eyys · 2022-09-28T14:34:12Z

Can you move e2e IT to seatunnel-connector-v2-e2e module? you can see #2924.

Can I submit this first? The branches are a bit messy

eyys · 2022-10-07T03:41:31Z

@ashulin @hailin0 Can you help review it

Hisoka-X · 2022-10-11T03:52:36Z

Please make sure CI passed first, thanks.

hailin0

LGTM

hailin0 · 2022-10-13T06:43:09Z

...nk-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/kafka/KafkaSourceJsonToConsoleIT.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")));

#3028

hailin0 · 2022-10-13T06:43:41Z

...nk-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/kafka/KafkaSourceTextToConsoleIT.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-10-13T06:44:03Z

...rk-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/kafka/KafkaSourceJsonToConsoleIT.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")));

...rk-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/kafka/KafkaSourceTextToConsoleIT.java

TyrantLucifer · 2022-10-16T12:50:46Z

...-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

+            }
+        } else {
+            typeInfo = SeaTunnelSchema.buildSimpleTextSchema();
+            this.deserializationSchema = null;


Suggested change

this.deserializationSchema = null;

this.deserializationSchema = TextDeserializationSchema.builder()

.seaTunnelRowType(typeInfo)

.delimiter(String.valueOf('\002'))

.build();

If user does not assign schema option, so connector will treat data as the following shown:

content

xxxxxxxx

So the upstream data does not need to be delimited, so just pass in an impossible delimiter when initialization the TextDeserializationSchema.

TyrantLucifer · 2022-10-16T12:55:31Z

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

+                                if (deserializationSchema != null) {
+                                    deserializationSchema.deserialize(record.value(), output);
+                                } else {
+                                    String content = stringDeserializer.deserialize(partition.topic(), record.value());
+                                    output.collect(new SeaTunnelRow(new Object[]{content}));
+                                }


Suggested change

if (deserializationSchema != null) {

deserializationSchema.deserialize(record.value(), output);

} else {

String content = stringDeserializer.deserialize(partition.topic(), record.value());

output.collect(new SeaTunnelRow(new Object[]{content}));

}

deserializationSchema.deserialize(record.value(), output)

TyrantLucifer

LGTM, thank you for your contribution. Let's waiting CI.

Hisoka-X

LGTM, waiting CI

TyrantLucifer

LGTM, Thank you for your contribution. CC @Hisoka-X

hailin0 reviewed Sep 19, 2022

View reviewed changes

eyys requested review from hailin0 and removed request for hailin0 September 20, 2022 14:29

eyys force-pushed the develop2 branch 2 times, most recently from c4321a4 to 7f30f2a Compare September 25, 2022 15:38

ashulin changed the title ~~[Feature][seatunnel-connectors-v2][connector-kafka] Kafka supports custom schema #2371~~ [Feature][connectors-v2][kafka] Kafka supports custom schema #2371 Sep 28, 2022

ashulin reviewed Sep 28, 2022

View reviewed changes

eyys force-pushed the develop2 branch 4 times, most recently from 87c53d3 to aef883c Compare October 7, 2022 00:23

eyys force-pushed the develop2 branch 5 times, most recently from 0a8c1b3 to 9e804ca Compare October 10, 2022 14:43

eyys force-pushed the develop2 branch 2 times, most recently from 6cdeaa6 to 0dafb56 Compare October 12, 2022 12:29

hailin0 previously approved these changes Oct 13, 2022

View reviewed changes

eyys dismissed hailin0’s stale review via ff56113 October 13, 2022 12:25

eyys force-pushed the develop2 branch 6 times, most recently from 532fcd8 to ba4b6c4 Compare October 16, 2022 08:38

TyrantLucifer reviewed Oct 16, 2022

View reviewed changes

eyys force-pushed the develop2 branch from ba4b6c4 to 8aaa809 Compare October 17, 2022 15:57

TyrantLucifer previously approved these changes Oct 17, 2022

View reviewed changes

Hisoka-X dismissed TyrantLucifer’s stale review via 9265e26 October 18, 2022 02:51

Hisoka-X previously approved these changes Oct 18, 2022

View reviewed changes

eyys dismissed Hisoka-X’s stale review via f4383f2 October 18, 2022 12:31

eyys force-pushed the develop2 branch from fff1287 to f4383f2 Compare October 18, 2022 12:31

[Feature][connectors-v2][kafka] Kafka supports custom schema apache#2371

6e132e0

eyys force-pushed the develop2 branch from f4383f2 to 6e132e0 Compare October 18, 2022 14:15

TyrantLucifer approved these changes Oct 19, 2022

View reviewed changes

Hisoka-X approved these changes Oct 19, 2022

View reviewed changes

Hisoka-X merged commit 6506e30 into apache:dev Oct 19, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][connectors-v2][kafka] Kafka supports custom schema #2371 #2783

[Feature][connectors-v2][kafka] Kafka supports custom schema #2371 #2783

eyys commented Sep 18, 2022

hailin0 Sep 19, 2022

eyys Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

eyys Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 Sep 19, 2022

hailin0 commented Sep 24, 2022

ashulin left a comment

eyys commented Sep 28, 2022

eyys commented Oct 7, 2022

Hisoka-X commented Oct 11, 2022

hailin0 left a comment

hailin0 Oct 13, 2022

hailin0 Oct 13, 2022

hailin0 Oct 13, 2022

TyrantLucifer Oct 16, 2022

TyrantLucifer Oct 16, 2022

TyrantLucifer left a comment

Hisoka-X left a comment

TyrantLucifer left a comment •

edited

Loading

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

-            this.deserializationSchema = null;
+            this.deserializationSchema = TextDeserializationSchema.builder()
+                    .seaTunnelRowType(typeInfo)
+                    .delimiter(String.valueOf('\002'))
+                    .build();

[Feature][connectors-v2][kafka] Kafka supports custom schema #2371 #2783

[Feature][connectors-v2][kafka] Kafka supports custom schema #2371 #2783

Conversation

eyys commented Sep 18, 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

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 commented Sep 24, 2022

ashulin left a comment

Choose a reason for hiding this comment

eyys commented Sep 28, 2022

eyys commented Oct 7, 2022

Hisoka-X commented Oct 11, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TyrantLucifer left a comment

Choose a reason for hiding this comment

Hisoka-X left a comment

Choose a reason for hiding this comment

TyrantLucifer left a comment • edited Loading

Choose a reason for hiding this comment

TyrantLucifer left a comment •

edited

Loading