fixed e2e test

apache · Jul 25, 2023 · a9e46e9 · a9e46e9
1 parent b509e54
commit a9e46e9
Show file tree

Hide file tree

Showing 10 changed files with 191 additions and 97 deletions.
diff --git a/seatunnel-connectors-v2/connector-kafka/pom.xml b/seatunnel-connectors-v2/connector-kafka/pom.xml
@@ -31,6 +31,7 @@
 
     <properties>
         <kafka.client.version>3.2.0</kafka.client.version>
+        <debezium.version>1.6.4.Final</debezium.version>
     </properties>
 
     <dependencies>
@@ -66,6 +67,28 @@
             <artifactId>seatunnel-format-compatible-connect-json</artifactId>
             <version>${project.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-embedded</artifactId>
+            <version>${debezium.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.kafka</groupId>
+                    <artifactId>kafka-log4j-appender</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.glassfish.jersey.core</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-mongodb</artifactId>
+            <version>${debezium.version}</version>
+        </dependency>
     </dependencies>
 
 </project>
diff --git a/...fka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java b/...fka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java
@@ -268,6 +268,7 @@ private void setDeserialization(Config config) {
                             CanalJsonDeserializationSchema.builder(typeInfo)
                                     .setIgnoreParseErrors(true)
                                     .build();
+                    break;
                 case COMPATIBLE_KAFKA_CONNECT_JSON:
                     deserializationSchema =
                             new CompatibleKafkaConnectDeserializationSchema(

diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/pom.xml
@@ -92,6 +92,11 @@
             <artifactId>postgresql</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>mysql</artifactId>

diff --git a/...fka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaConnectToKafkaIT.java b/...fka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaConnectToKafkaIT.java
diff --git a/...l-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/ddl/seatunnel.sql b/...l-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/ddl/seatunnel.sql
diff --git a/...r-v2-e2e/connector-kafka-e2e/src/test/resources/kafkasource_debezium_record_to_mysql.conf b/...r-v2-e2e/connector-kafka-e2e/src/test/resources/kafkasource_debezium_record_to_mysql.conf
@@ -32,7 +32,7 @@ env {
 
 source {
   Kafka {
-    bootstrap.servers = "kafkaCluster:9092"
+    bootstrap.servers = "kafka_connect_source_record:9092"
     topic = "debezium_source_record"
     result_table_name = "kafka_table"
     start_mode = earliest
@@ -52,8 +52,8 @@ source {
 
 sink {
     Jdbc {
-        driver = com.mysql.cj.jdbc.Driver
-        url = "jdbc:mysql://mysql_e2e:3306/seatunnel?useSSL=false"
+        driver = "com.mysql.cj.jdbc.Driver"
+        url = "jdbc:mysql://kafka_to_mysql_e2e:3306/seatunnel"
         user = st_user
         password = seatunnel
         generate_sink_sql = true

diff --git a/...ector-v2-e2e/connector-kafka-e2e/src/test/resources/kafkasource_jdbc_record_to_mysql.conf b/...ector-v2-e2e/connector-kafka-e2e/src/test/resources/kafkasource_jdbc_record_to_mysql.conf
@@ -32,7 +32,7 @@ env {
 
 source {
   Kafka {
-    bootstrap.servers = "kafkaCluster:9092"
+    bootstrap.servers = "kafka_connect_source_record:9092"
     topic = "jdbc_source_record"
     result_table_name = "kafka_table"
     start_mode = earliest
@@ -53,7 +53,7 @@ source {
 sink {
     Jdbc {
         driver = com.mysql.cj.jdbc.Driver
-        url = "jdbc:mysql://mysql_e2e:3306/seatunnel?useSSL=false"
+        url = "jdbc:mysql://kafka_to_mysql_e2e:3306/seatunnel"
         user = st_user
         password = seatunnel
         generate_sink_sql = true

diff --git a/seatunnel-formats/seatunnel-format-compatible-connect-json/pom.xml b/seatunnel-formats/seatunnel-format-compatible-connect-json/pom.xml
@@ -1,4 +1,18 @@
 <?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+-->
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
@@ -11,8 +25,7 @@
     <artifactId>seatunnel-format-compatible-connect-json</artifactId>
     <name>SeaTunnel : Formats : Compatible Kafka Connect Json</name>
     <properties>
-        <kafka.connect.version>3.4.0</kafka.connect.version>
-        <debezium.version>2.1.2.Final</debezium.version>
+        <debezium.version>1.6.4.Final</debezium.version>
     </properties>
 
     <dependencies>
@@ -24,34 +37,34 @@
         </dependency>
 
         <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>connect-json</artifactId>
-            <version>${kafka.connect.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>connect-runtime</artifactId>
-            <version>${kafka.connect.version}</version>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
         </dependency>
 
         <dependency>
             <groupId>io.debezium</groupId>
-            <artifactId>debezium-core</artifactId>
+            <artifactId>debezium-embedded</artifactId>
             <version>${debezium.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.seatunnel</groupId>
-            <artifactId>seatunnel-format-json</artifactId>
-            <version>${project.version}</version>
-            <scope>compile</scope>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.kafka</groupId>
+                    <artifactId>kafka-log4j-appender</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.glassfish.jersey.core</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
             <groupId>io.debezium</groupId>
             <artifactId>debezium-connector-mongodb</artifactId>
             <version>${debezium.version}</version>
+            <scope>provided</scope>
         </dependency>
 
     </dependencies>

diff --git a/...nel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java b/...nel/format/compatible/kafka/connect/json/CompatibleKafkaConnectDeserializationSchema.java
@@ -35,11 +35,11 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.header.Header;
 import org.apache.kafka.connect.json.JsonConverter;
 import org.apache.kafka.connect.json.JsonConverterConfig;
 import org.apache.kafka.connect.sink.SinkRecord;
-import org.apache.kafka.connect.util.ConnectUtils;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -183,10 +183,11 @@ private SeaTunnelRow convertJsonNode(JsonNode jsonNode) {
 
     private SinkRecord convertToSinkRecord(ConsumerRecord<byte[], byte[]> msg) {
         SchemaAndValue keyAndSchema =
-                keyConverter.toConnectData(msg.topic(), msg.headers(), msg.key());
+                (msg.key() == null)
+                        ? SchemaAndValue.NULL
+                        : keyConverter.toConnectData(msg.topic(), msg.headers(), msg.key());
         SchemaAndValue valueAndSchema =
                 valueConverter.toConnectData(msg.topic(), msg.headers(), msg.value());
-        Long timestamp = ConnectUtils.checkAndConvertTimestamp(msg.timestamp());
         return new SinkRecord(
                 msg.topic(),
                 msg.partition(),
@@ -195,22 +196,28 @@ private SinkRecord convertToSinkRecord(ConsumerRecord<byte[], byte[]> msg) {
                 valueAndSchema.schema(),
                 valueAndSchema.value(),
                 msg.offset(),
-                timestamp,
+                msg.timestamp(),
                 msg.timestampType(),
                 null);
     }
 
     private SinkRecord convertByTransforms(SinkRecord record) {
+        tryInitTransforms();
+        String connector;
         Header header = record.headers().lastWithName(DBZ_SOURCE_CONNECTOR);
         if (header != null) {
-            String connector = header.value().toString();
-            if (DBZ_SOURCE_CONNECTOR_MONGODB_VALUE.equals(connector)) {
-                return extractNewDocumentState.apply(record);
-            } else {
-                return this.extractNewRecordState.apply(record);
-            }
+            connector = header.value().toString();
+        } else {
+            connector =
+                    ((Struct) (((Struct) record.value()).get("source")))
+                            .get("connector")
+                            .toString();
+        }
+        if (DBZ_SOURCE_CONNECTOR_MONGODB_VALUE.equals(connector)) {
+            return extractNewDocumentState.apply(record);
+        } else {
+            return this.extractNewRecordState.apply(record);
         }
-        return record;
     }
 
     @Override

diff --git a/seatunnel-formats/seatunnel-format-json/pom.xml b/seatunnel-formats/seatunnel-format-json/pom.xml
@@ -35,12 +35,6 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.kafka</groupId>
-            <artifactId>connect-json</artifactId>
-            <version>3.4.0</version>
-            <scope>compile</scope>
-        </dependency>
 
     </dependencies>