[improve][io] Remove kafka-connect-avro-converter-shaded (apache#19468)

Signed-off-by: tison <[email protected]>
eugene-cheverda · Feb 9, 2023 · b969fe5 · b969fe5
1 parent 329c8c0
commit b969fe5
Show file tree

Hide file tree

Showing 9 changed files with 16 additions and 154 deletions.
diff --git a/build/run_unit_group.sh b/build/run_unit_group.sh
@@ -179,7 +179,6 @@ function test_group_other() {
 }
 
 function test_group_pulsar_io() {
-    $MVN_TEST_OPTIONS -pl kafka-connect-avro-converter-shaded clean install
     echo "::group::Running pulsar-io tests"
     mvn_test --install -Ppulsar-io-tests,-main
     echo "::endgroup::"

diff --git a/kafka-connect-avro-converter-shaded/pom.xml b/kafka-connect-avro-converter-shaded/pom.xml
diff --git a/pom.xml b/pom.xml
@@ -2149,9 +2149,6 @@ flexible messaging model and an intuitive client API.</description>
         <!-- connector-related modules -->
         <module>pulsar-io</module>
 
-        <!-- kafka connect avro converter shaded, because version mismatch    -->
-        <module>kafka-connect-avro-converter-shaded</module>
-
         <!-- Bouncy Castle Provider loaders-->
         <module>bouncy-castle</module>
 

diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -104,25 +104,13 @@
       <artifactId>commons-lang3</artifactId>
     </dependency>
 
+    <!-- confluent connect avro converter -->
     <dependency>
-      <groupId>${project.groupId}</groupId>
-      <artifactId>kafka-connect-avro-converter-shaded</artifactId>
-      <version>${project.version}</version>
-      <exclusions>
-        <exclusion>
-          <groupId>io.confluent</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-        <exclusion>
-          <groupId>org.apache.avro</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-      </exclusions>
+      <groupId>io.confluent</groupId>
+      <artifactId>kafka-connect-avro-converter</artifactId>
+      <version>${confluent.version}</version>
     </dependency>
 
-
-
-
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>pulsar-broker</artifactId>

diff --git a/...-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java b/...-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/AbstractKafkaConnectSource.java
@@ -18,6 +18,9 @@
  */
 package org.apache.pulsar.io.kafka.connect;
 
+import io.confluent.connect.avro.AvroConverter;
+import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
+import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -43,9 +46,6 @@
 import org.apache.pulsar.io.core.Source;
 import org.apache.pulsar.io.core.SourceContext;
 import org.apache.pulsar.io.kafka.connect.schema.KafkaSchemaWrappedSchema;
-import org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter;
-import org.apache.pulsar.kafka.shade.io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
-import org.apache.pulsar.kafka.shade.io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
 
 /**
  * A pulsar source that runs.

diff --git a/...-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java b/...-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
@@ -20,6 +20,7 @@
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
+import io.confluent.connect.avro.AvroData;
 import java.util.Base64;
 import java.util.Map;
 import java.util.Optional;
@@ -34,7 +35,6 @@
 import org.apache.pulsar.functions.api.KVRecord;
 import org.apache.pulsar.io.core.SourceContext;
 import org.apache.pulsar.io.kafka.connect.schema.KafkaSchemaWrappedSchema;
-import org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroData;
 
 /**
  * A pulsar source that runs.

diff --git a/...tor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/KafkaSchemaWrappedSchema.java b/...tor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/KafkaSchemaWrappedSchema.java
@@ -36,10 +36,9 @@
 @Slf4j
 public class KafkaSchemaWrappedSchema implements Schema<byte[]>, Serializable {
 
-    private SchemaInfo schemaInfo = null;
+    private final SchemaInfo schemaInfo;
 
-    public KafkaSchemaWrappedSchema(org.apache.pulsar.kafka.shade.avro.Schema schema,
-                                    Converter converter) {
+    public KafkaSchemaWrappedSchema(org.apache.avro.Schema schema, Converter converter) {
         Map<String, String> props = new HashMap<>();
         boolean isJsonConverter = converter instanceof JsonConverter;
         props.put(GenericAvroSchema.OFFSET_PROP, isJsonConverter ? "0" : "5");

diff --git a/...or/src/main/java/org/apache/pulsar/io/kafka/connect/schema/PulsarSchemaToKafkaSchema.java b/...or/src/main/java/org/apache/pulsar/io/kafka/connect/schema/PulsarSchemaToKafkaSchema.java
@@ -24,6 +24,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.util.concurrent.ExecutionError;
 import com.google.common.util.concurrent.UncheckedExecutionException;
+import io.confluent.connect.avro.AvroData;
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -37,7 +38,6 @@
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.pulsar.client.api.schema.KeyValueSchema;
 import org.apache.pulsar.common.schema.SchemaType;
-import org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroData;
 
 @Slf4j
 public class PulsarSchemaToKafkaSchema {
@@ -74,9 +74,8 @@ public static boolean matchesToKafkaLogicalSchema(Schema kafkaSchema) {
     }
 
     // Parse json to shaded schema
-    private static org.apache.pulsar.kafka.shade.avro.Schema parseAvroSchema(String schemaJson) {
-        final org.apache.pulsar.kafka.shade.avro.Schema.Parser parser =
-                new org.apache.pulsar.kafka.shade.avro.Schema.Parser();
+    private static org.apache.avro.Schema parseAvroSchema(String schemaJson) {
+        final org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser();
         parser.setValidateDefaults(false);
         return parser.parse(schemaJson);
     }
@@ -126,9 +125,8 @@ public static Schema getKafkaConnectSchema(org.apache.pulsar.client.api.Schema p
                                              getKafkaConnectSchema(kvSchema.getValueSchema()))
                                 .build();
                 }
-                org.apache.pulsar.kafka.shade.avro.Schema avroSchema =
-                        parseAvroSchema(new String(pulsarSchema.getSchemaInfo().getSchema(),
-                                StandardCharsets.UTF_8));
+                org.apache.avro.Schema avroSchema = parseAvroSchema(
+                        new String(pulsarSchema.getSchemaInfo().getSchema(), StandardCharsets.UTF_8));
                 return avroData.toConnectSchema(avroSchema);
             });
         } catch (ExecutionException | UncheckedExecutionException | ExecutionError ee) {

diff --git a/...va/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java b/...va/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java
@@ -60,8 +60,7 @@ public void testDebeziumMySqlSourceJsonWithClientBuilder() throws Exception {
 
     @Test(groups = "source")
     public void testDebeziumMySqlSourceAvro() throws Exception {
-        testDebeziumMySqlConnect(
-                "org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter", false, false);
+        testDebeziumMySqlConnect("io.confluent.connect.avro.AvroConverter", false, false);
     }
 
     @Test(groups = "source")