Issue/function avro schema decode error new

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1522,8 +1522,8 @@ protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGet
 
     @Override
     protected void handleGetSchema(CommandGetSchema commandGetSchema) {
-        if (log.isDebugEnabled()) {
-            log.debug("Received CommandGetSchema call from {}", remoteAddress);
+        if (log.isInfoEnabled()) {
+            log.info("Received CommandGetSchema call from {}", remoteAddress);
         }
 
         long requestId = commandGetSchema.getRequestId();
@@ -1558,8 +1558,8 @@ protected void handleGetSchema(CommandGetSchema commandGetSchema) {
 
     @Override
     protected void handleGetOrCreateSchema(CommandGetOrCreateSchema commandGetOrCreateSchema) {
-        if (log.isDebugEnabled()) {
-            log.debug("Received CommandGetOrCreateSchema call from {}", remoteAddress);
+        if (log.isInfoEnabled()) {
+            log.info("Received CommandGetOrCreateSchema call from {}", remoteAddress);
         }
         long requestId = commandGetOrCreateSchema.getRequestId();
         String topicName = commandGetOrCreateSchema.getTopic();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java

@@ -371,6 +371,7 @@ static List<SchemaRegistryFormat.SchemaInfo.KeyValuePair> toPairs(Map<String, St
         }
 
         static SchemaData schemaInfoToSchema(SchemaRegistryFormat.SchemaInfo info) {
+            log.info("SchemaInfo to shema : info type {}", info.getType());
             return SchemaData.builder()
                 .user(info.getUser())
                 .type(convertToDomainType(info.getType()))

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

@@ -65,9 +65,12 @@ public class AvroSchema<T> extends StructSchema<T> {
         reflectDataNotAllowNull.addLogicalTypeConversion(new TimeConversions.TimestampMicrosConversion());
     }
 
-    private AvroSchema(SchemaInfo schemaInfo) {
+    private ClassLoader pojoClassLoader;
+
+    private AvroSchema(SchemaInfo schemaInfo, ClassLoader pojoClassLoader) {
         super(schemaInfo);
-        setReader(new AvroReader<>(schema));
+        this.pojoClassLoader = pojoClassLoader;
+        setReader(new AvroReader<>(schema, pojoClassLoader));
         setWriter(new AvroWriter<>(schema));
     }
 
@@ -78,34 +81,43 @@ public boolean supportSchemaVersioning() {
 
     @Override
     public Schema<T> clone() {
-        Schema<T> schema = new AvroSchema<>(schemaInfo);
+        Schema<T> schema = new AvroSchema<>(schemaInfo, pojoClassLoader);
         if (schemaInfoProvider != null) {
             schema.setSchemaInfoProvider(schemaInfoProvider);
         }
         return schema;
     }
 
     public static <T> AvroSchema<T> of(SchemaDefinition<T> schemaDefinition) {
-        return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO));
+        ClassLoader pojoClassLoader = null;
+        if (schemaDefinition.getPojo() != null) {
+            pojoClassLoader = schemaDefinition.getPojo().getClassLoader();
+            log.info("pojo: {}, pojoClassLoader: {}", schemaDefinition.getPojo().getName(), pojoClassLoader);
+        }
+        return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO), pojoClassLoader);
     }
 
     public static <T> AvroSchema<T> of(Class<T> pojo) {
         return AvroSchema.of(SchemaDefinition.<T>builder().withPojo(pojo).build());
     }
 
     public static <T> AvroSchema<T> of(Class<T> pojo, Map<String, String> properties) {
+        ClassLoader pojoClassLoader = null;
+        if (pojo != null) {
+            pojoClassLoader = pojo.getClassLoader();
+        }
         SchemaDefinition<T> schemaDefinition = SchemaDefinition.<T>builder().withPojo(pojo).withProperties(properties).build();
-        return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO));
+        return new AvroSchema<>(parseSchemaInfo(schemaDefinition, SchemaType.AVRO), pojoClassLoader);
     }
 
     @Override
     protected SchemaReader<T> loadReader(BytesSchemaVersion schemaVersion) {
         SchemaInfo schemaInfo = getSchemaInfoByVersion(schemaVersion.get());
         if (schemaInfo != null) {
-            log.info("Load schema reader for version({}), schema is : {}",
+            log.info("Load schema reader for version({}), schema is : {}, schemaInfo: {}",
                 SchemaUtils.getStringSchemaVersion(schemaVersion.get()),
-                schemaInfo.getSchemaDefinition());
-            return new AvroReader<>(parseAvroSchema(schemaInfo.getSchemaDefinition()), schema);
+                schemaInfo.getSchemaDefinition(), schemaInfo.toString());
+            return new AvroReader<>(parseAvroSchema(schemaInfo.getSchemaDefinition()), schema, pojoClassLoader);
         } else {
             log.warn("No schema found for version({}), use latest schema : {}",
                 SchemaUtils.getStringSchemaVersion(schemaVersion.get()),

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

@@ -18,9 +18,12 @@
  */
 package org.apache.pulsar.client.impl.schema.reader;
 
+import org.apache.avro.Conversions;
 import org.apache.avro.Schema;
+import org.apache.avro.data.TimeConversions;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.reflect.ReflectData;
 import org.apache.avro.reflect.ReflectDatumReader;
 import org.apache.pulsar.client.api.SchemaSerializationException;
 import org.apache.pulsar.client.api.schema.SchemaReader;
@@ -41,8 +44,34 @@ public AvroReader(Schema schema) {
         this.reader = new ReflectDatumReader<>(schema);
     }
 
-    public AvroReader(Schema writerSchema, Schema readerSchema) {
-        this.reader = new ReflectDatumReader<>(writerSchema, readerSchema);
+    public AvroReader(Schema schema, ClassLoader classLoader) {
+        if (classLoader != null) {
+            ReflectData reflectData = new ReflectData(classLoader);
+            reflectData.addLogicalTypeConversion(new Conversions.DecimalConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.DateConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimeMillisConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimeMicrosConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMillisConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMicrosConversion());
+            this.reader = new ReflectDatumReader<>(schema, schema, reflectData);
+        } else {
+            this.reader = new ReflectDatumReader<>(schema);
+        }
+    }
+
+    public AvroReader(Schema writerSchema, Schema readerSchema, ClassLoader classLoader) {
+        if (classLoader != null) {
+            ReflectData reflectData = new ReflectData(classLoader);
+            reflectData.addLogicalTypeConversion(new Conversions.DecimalConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.DateConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimeMillisConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimeMicrosConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMillisConversion());
+            reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMicrosConversion());
+            this.reader = new ReflectDatumReader<>(writerSchema, readerSchema, reflectData);
+        } else {
+            this.reader = new ReflectDatumReader<>(writerSchema, readerSchema);
+        }
     }
 
     @Override

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

@@ -721,6 +721,7 @@ public static SchemaType getSchemaType(Schema.Type type) {
     }
 
     private static Schema getSchema(SchemaInfo schemaInfo) {
+        log.info("Schema info converted to protobuf schema : " + schemaInfo);
         Schema.Builder builder = Schema.newBuilder()
             .setName(schemaInfo.getName())
             .setSchemaData(copyFrom(schemaInfo.getSchema()))

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaInfoUtil.java

@@ -38,6 +38,8 @@ public static SchemaInfo newSchemaInfo(String name, SchemaData data) {
         si.setSchema(data.getData());
         si.setType(data.getType());
         si.setProperties(data.getProps());
+        System.out.println("newSchemaInfo: schema name = " + name
+                + ", type = " + data.getType() + ": " + si);
         return si;
     }
 
@@ -46,6 +48,8 @@ public static SchemaInfo newSchemaInfo(Schema schema) {
         si.setName(schema.getName());
         si.setSchema(schema.getSchemaData().toByteArray());
         si.setType(Commands.getSchemaType(schema.getType()));
+        System.out.println("Coverting schema '" + schema.getName()
+                + "' type from " + schema.getType() + " to " + si.getType());
         if (schema.getPropertiesCount() == 0) {
             si.setProperties(Collections.emptyMap());
         } else {

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

@@ -317,8 +317,8 @@ private ClassLoader loadJars() throws Exception {
                     Collections.emptyList());
         }
 
-        log.info("Initialize function class loader for function {} at function cache manager",
-                instanceConfig.getFunctionDetails().getName());
+        log.info("Initialize function class loader for function {} at function cache manager, functionClassLoader: {}",
+                instanceConfig.getFunctionDetails().getName(), fnCache.getClassLoader(instanceConfig.getFunctionId()));
 
         fnClassLoader = fnCache.getClassLoader(instanceConfig.getFunctionId());
         if (null == fnClassLoader) {

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java

@@ -97,6 +97,7 @@ private SchemaType getSchemaTypeOrDefault(String topic, Class<?> clazz) {
         } else {
             Optional<SchemaInfo> schema = ((PulsarClientImpl) client).getSchema(topic).join();
             if (schema.isPresent()) {
+                System.out.println("[getSchemaTypeOrDefault] topic : " + topic + ": schema info : " + schema.get());
                 if (schema.get().getType() == SchemaType.NONE) {
                     return getDefaultSchemaType(clazz);
                 } else {

pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/AvroSchemaTestFunction.java

@@ -0,0 +1,36 @@
+/**
+ * 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.examples;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Context;
+import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.api.examples.pojo.AvroTestObject;
+
+
+@Slf4j
+public class AvroSchemaTestFunction implements Function<AvroTestObject, AvroTestObject> {
+
+    @Override
+    public AvroTestObject process(AvroTestObject input, Context context) throws Exception {
+        log.info("AvroTestObject - baseValue: {}", input.getBaseValue());
+        input.setBaseValue(input.getBaseValue() + 10);
+        return input;
+    }
+}

pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/pojo/AvroTestObject.java

@@ -0,0 +1,32 @@
+/**
+ * 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.examples.pojo;
+
+import lombok.Data;
+
+
+/**
+ * Avro test object.
+ */
+@Data
+public class AvroTestObject {
+
+    private int baseValue;
+
+}

tests/docker-images/latest-version-image/conf/bookie.conf

@@ -22,5 +22,5 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/bookie.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M",PULSAR_GC="-XX:+UseG1GC",dbStorage_writeCacheMaxSizeMb="16",dbStorage_readAheadCacheMaxSizeMb="16"
+environment=PULSAR_MEM="-Xmx128M -XX:MaxDirectMemorySize=512M",PULSAR_GC="-XX:+UseG1GC",dbStorage_writeCacheMaxSizeMb="16",dbStorage_readAheadCacheMaxSizeMb="16"
 command=/pulsar/bin/pulsar bookie

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java

@@ -139,4 +139,29 @@ public int hashCode() {
             clusterName);
     }
 
+    @Override
+    public void start() {
+        super.start();
+        this.tailContainerLog();
+        if (this.getContainerName().contains("pulsar-broker")) {
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "tail", "-f", "/var/log/pulsar/broker.log");
+        } else if (this.getContainerName().contains("bookie")) {
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "tail", "-f", "/var/log/pulsar/bookie.log");
+        } else if (this.getContainerName().contains("functions-worker")) {
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "tail", "-f", "/var/log/pulsar/functions_worker.log");
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "mkdir", "-p",
+                    "/tmp/functions/public/default/test-avroschema-fn-202003241756");
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "touch",
+                    "/tmp/functions/public/default/test-avroschema-fn-202003241756/test-avroschema-fn-202003241756-0.log");
+            DockerUtils.runCommandAsync(this.dockerClient, this.getContainerId(),
+                    "tail", "-f",
+                    "/tmp/functions/public/default/test-avroschema-fn-202003241756/test-avroschema-fn-202003241756-0.log");
+        }
+    }
+
 }