[Pulsar IO][Issue 5633]Support avro schema for debezium connector

kafka-connect-avro-converter-shaded/pom.xml

@@ -0,0 +1,128 @@
+<?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>
+    <parent>
+        <artifactId>pulsar</artifactId>
+        <groupId>org.apache.pulsar</groupId>
+        <version>2.6.0-SNAPSHOT</version>
+        <relativePath>..</relativePath>
+    </parent>
+
+    <artifactId>kafka-connect-avro-converter-shaded</artifactId>
+    <name>Apache Pulsar :: Kafka Connect Avro Converter shaded</name>
+
+    <dependencies>
+        <!-- confluent connect avro converter -->
+        <dependency>
+            <groupId>io.confluent</groupId>
+            <artifactId>kafka-connect-avro-converter</artifactId>
+            <version>${confluent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-core-asl</artifactId>
+            <version>${kafka-avro-convert-jackson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.codehaus.jackson</groupId>
+            <artifactId>jackson-mapper-asl</artifactId>
+            <version>${kafka-avro-convert-jackson.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <createDependencyReducedPom>true</createDependencyReducedPom>
+                    <promoteTransitiveDependencies>true</promoteTransitiveDependencies>
+
+                    <artifactSet>
+                        <includes>
+                            <include>io.confluent:*</include>
+                            <include>io.confluent:kafka-avro-serializer</include>
+                            <include>io.confluent:kafka-schema-registry-client</include>
+                            <include>io.confluent:common-config</include>
+                            <include>io.confluent:common-utils</include>
+                            <include>org.apache.avro:*</include>
+
+                            <include>org.codehaus.jackson:jackson-core-asl</include>
+                            <include>org.codehaus.jackson:jackson-mapper-asl</include>
+                            <include>com.thoughtworks.paranamer:paranamer</include>
+                            <include>org.xerial.snappy:snappy-java</include>
+                            <include>org.apache.commons:commons-compress</include>
+                            <include>org.tukaani:xz</include>
+                        </includes>
+                    </artifactSet>
+                    <relocations>
+                        <relocation>
+                            <pattern>io.confluent</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.io.confluent</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>org.apache.avro</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.avro</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>org.codehaus.jackson</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.org.codehaus.jackson</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>com.thoughtworks.paranamer</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.com.thoughtworks.paranamer</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>org.xerial.snappy</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.org.xerial.snappy</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>org.apache.commons</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.org.apache.commons</shadedPattern>
+                        </relocation>
+                        <relocation>
+                            <pattern>org.tukaani</pattern>
+                            <shadedPattern>org.apache.pulsar.kafka.shade.org.tukaani</shadedPattern>
+                        </relocation>
+                    </relocations>
+                    <transformers>
+                        <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+                        <transformer implementation="org.apache.maven.plugins.shade.resource.PluginXmlResourceTransformer" />
+                    </transformers>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

pom.xml

@@ -117,6 +117,9 @@ 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>
+
     <!-- examples -->
     <module>examples</module>
 
@@ -208,6 +211,8 @@ flexible messaging model and an intuitive client API.</description>
     <guava.version>25.1-jre</guava.version>
     <jcip.version>1.0</jcip.version>
     <prometheus-jmx.version>0.12.0</prometheus-jmx.version>
+    <confluent.version>5.3.2</confluent.version>
+    <kafka-avro-convert-jackson.version>1.9.13</kafka-avro-convert-jackson.version>
 
     <!-- test dependencies -->
     <cassandra.version>3.6.0</cassandra.version>
@@ -220,6 +225,7 @@ flexible messaging model and an intuitive client API.</description>
     <javassist.version>3.25.0-GA</javassist.version>
     <failsafe.version>2.3.1</failsafe.version>
     <skyscreamer.version>1.5.0</skyscreamer.version>
+    <confluent.version>5.2.2</confluent.version>
 
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
@@ -1715,5 +1721,9 @@ flexible messaging model and an intuitive client API.</description>
       <id>spring-plugins-release</id>
       <url>https://repo.spring.io/plugins-release/</url>
     </repository>
+    <repository>
+      <id>confluent</id>
+      <url>http://packages.confluent.io/maven/</url>
+    </repository>
   </repositories>
 </project>

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java

@@ -41,6 +41,7 @@
 import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
 import org.apache.pulsar.client.api.schema.SchemaReader;
 import org.apache.pulsar.client.api.schema.SchemaWriter;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
 import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -78,6 +79,11 @@ public SchemaReader<T> load(BytesSchemaVersion schemaVersion) {
     protected StructSchema(SchemaInfo schemaInfo) {
         this.schema = parseAvroSchema(new String(schemaInfo.getSchema(), UTF_8));
         this.schemaInfo = schemaInfo;
+
+        if (schemaInfo.getProperties().containsKey(GenericAvroSchema.OFFSET_PROP)) {
+            this.schema.addProp(GenericAvroSchema.OFFSET_PROP,
+                    schemaInfo.getProperties().get(GenericAvroSchema.OFFSET_PROP));
+        }
     }
 
     public org.apache.avro.Schema getAvroSchema() {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReader.java

@@ -47,6 +47,8 @@ public class GenericAvroReader implements SchemaReader<GenericRecord> {
     private final List<Field> fields;
     private final Schema schema;
     private final byte[] schemaVersion;
+    private int offset;
+
     public GenericAvroReader(Schema schema) {
         this(null, schema, null);
     }
@@ -65,12 +67,22 @@ public GenericAvroReader(Schema writerSchema, Schema readerSchema, byte[] schema
         }
         this.byteArrayOutputStream = new ByteArrayOutputStream();
         this.encoder = EncoderFactory.get().binaryEncoder(this.byteArrayOutputStream, encoder);
+
+        if (schema.getObjectProp(GenericAvroSchema.OFFSET_PROP) != null) {
+            this.offset = Integer.parseInt(schema.getObjectProp(GenericAvroSchema.OFFSET_PROP).toString());
+        } else {
+            this.offset = 0;
+        }
+
     }
 
     @Override
     public GenericAvroRecord read(byte[] bytes, int offset, int length) {
         try {
-            Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length, null);
+            if (offset == 0 && this.offset > 0) {
+                offset = this.offset;
+            }
+            Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length - offset, null);
             org.apache.avro.generic.GenericRecord avroRecord =
                     (org.apache.avro.generic.GenericRecord)reader.read(
                     null,
@@ -101,5 +113,9 @@ public GenericRecord read(InputStream inputStream) {
         }
     }
 
+    public int getOffset() {
+        return offset;
+    }
+
     private static final Logger log = LoggerFactory.getLogger(GenericAvroReader.class);
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java

@@ -33,6 +33,8 @@
 @Slf4j
 public class GenericAvroSchema extends GenericSchemaImpl {
 
+    public final static String OFFSET_PROP = "__AVRO_READ_OFFSET__";
+
     public GenericAvroSchema(SchemaInfo schemaInfo) {
         this(schemaInfo, true);
     }
@@ -73,6 +75,8 @@ protected SchemaReader<GenericRecord> loadReader(BytesSchemaVersion schemaVersio
                  schemaInfo);
              Schema writerSchema = parseAvroSchema(schemaInfo.getSchemaDefinition());
              Schema readerSchema = useProvidedSchemaAsReaderSchema ? schema : writerSchema;
+             readerSchema.addProp(OFFSET_PROP, schemaInfo.getProperties().getOrDefault(OFFSET_PROP, "0"));
+
              return new GenericAvroReader(
                      writerSchema,
                      readerSchema,

pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReaderTest.java

@@ -20,6 +20,8 @@
 
 import static org.testng.Assert.assertEquals;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
@@ -38,18 +40,22 @@ public class GenericAvroReaderTest {
     private AvroSchema fooSchemaNotNull;
     private AvroSchema fooSchema;
     private AvroSchema fooV2Schema;
-
+    private AvroSchema fooOffsetSchema;
 
     @BeforeMethod
     public void setup() {
         fooSchema = AvroSchema.of(Foo.class);
+
         fooV2Schema = AvroSchema.of(FooV2.class);
         fooSchemaNotNull = AvroSchema.of(SchemaDefinition
                 .builder()
                 .withAlwaysAllowNull(false)
                 .withPojo(Foo.class)
                 .build());
 
+        fooOffsetSchema = AvroSchema.of(Foo.class);
+        fooOffsetSchema.getAvroSchema().addProp(GenericAvroSchema.OFFSET_PROP, 5);
+
         foo = new Foo();
         foo.setField1("foo1");
         foo.setField2("bar1");
@@ -83,4 +89,20 @@ public void testGenericAvroReaderByReaderSchema() {
         assertEquals(genericRecordByReaderSchema.getField("field3"), 10);
     }
 
+    @Test
+    public void testOffsetSchema() {
+        byte[] fooBytes = fooOffsetSchema.encode(foo);
+        ByteBuf byteBuf = Unpooled.buffer();
+        byteBuf.writeByte(0);
+        byteBuf.writeInt(10);
+        byteBuf.writeBytes(fooBytes);
+
+        GenericAvroReader reader = new GenericAvroReader(fooOffsetSchema.getAvroSchema());
+        assertEquals(reader.getOffset(), 5);
+        GenericRecord record = reader.read(byteBuf.array());
+        assertEquals(record.getField("field1"), "foo1");
+        assertEquals(record.getField("field2"), "bar1");
+        assertEquals(record.getField("fieldUnableNull"), "notNull");
+    }
+
 }

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java

@@ -0,0 +1,35 @@
+/**
+ * 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.
+ */
+package org.apache.pulsar.functions.api;
+
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+
+/**
+ * key value schema record.
+ */
+public interface KVRecord<K, V> extends Record {
+
+    Schema<K> getKeySchema();
+
+    Schema<V> getValueSchema();
+
+    KeyValueEncodingType getKeyValueEncodingType();
+
+}

pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.functions.api;
 
 import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
 
 import java.util.Collections;
 import java.util.Map;
@@ -43,6 +44,10 @@ default Optional<String> getKey() {
         return Optional.empty();
     }
 
+    default Schema<T> getSchema() {
+        return null;
+    }
+
     /**
      * Retrieves the actual data of the record.
      *

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java

@@ -24,8 +24,13 @@
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.impl.schema.KeyValueSchema;
+import org.apache.pulsar.functions.api.KVRecord;
 import org.apache.pulsar.functions.api.Record;
 
+@Slf4j
 @Data
 @AllArgsConstructor
 public class SinkRecord<T> implements Record<T> {
@@ -81,4 +86,24 @@ public void fail() {
     public Optional<String> getDestinationTopic() {
         return sourceRecord.getDestinationTopic();
     }
+
+    @Override
+    public Schema<T> getSchema() {
+        if (sourceRecord == null) {
+            return null;
+        }
+
+        if (sourceRecord.getSchema() != null) {
+            return sourceRecord.getSchema();
+        }
+
+        if (sourceRecord instanceof KVRecord) {
+            KVRecord kvRecord = (KVRecord) sourceRecord;
+            return KeyValueSchema.of(kvRecord.getKeySchema(), kvRecord.getValueSchema(),
+                    kvRecord.getKeyValueEncodingType());
+        }
+
+        return null;
+    }
+
 }