Java Client: MessageImpl - ensure that AutoConsumeSchema downloaded the schema before decoding the payload

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

@@ -1707,9 +1707,15 @@ protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGet
     @Override
     protected void handleGetSchema(CommandGetSchema commandGetSchema) {
         if (log.isDebugEnabled()) {
-            log.debug("Received CommandGetSchema call from {}, schemaVersion: {}, topic: {}, requestId: {}",
-                    remoteAddress, new String(commandGetSchema.getSchemaVersion()),
-                    commandGetSchema.getTopic(), commandGetSchema.getRequestId());
+            if (commandGetSchema.hasSchemaVersion()) {
+                log.debug("Received CommandGetSchema call from {}, schemaVersion: {}, topic: {}, requestId: {}",
+                        remoteAddress, new String(commandGetSchema.getSchemaVersion()),
+                        commandGetSchema.getTopic(), commandGetSchema.getRequestId());
+            } else {
+                log.debug("Received CommandGetSchema call from {}, schemaVersion: {}, topic: {}, requestId: {}",
+                        remoteAddress, null,
+                        commandGetSchema.getTopic(), commandGetSchema.getRequestId());
+            }
         }
 
         long requestId = commandGetSchema.getRequestId();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java

@@ -50,6 +50,7 @@
 import org.apache.pulsar.common.api.proto.SingleMessageMetadata;
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 
 public class MessageImpl<T> implements Message<T> {
@@ -333,9 +334,17 @@ public byte[] getSchemaVersion() {
         }
     }
 
+    private SchemaInfo getSchemaInfo() {
+        if (schema instanceof AutoConsumeSchema) {
+            ((AutoConsumeSchema) schema).fetchSchemaIfNeeded();
+        }
+        return schema.getSchemaInfo();
+    }
+
     @Override
     public T getValue() {
-        if (schema.getSchemaInfo() != null && SchemaType.KEY_VALUE == schema.getSchemaInfo().getType()) {
+        SchemaInfo schemaInfo = getSchemaInfo();
+        if (schemaInfo != null && SchemaType.KEY_VALUE == schemaInfo.getType()) {
             if (schema.supportSchemaVersioning()) {
                 return getKeyValueBySchemaVersion();
             } else {

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

@@ -21,7 +21,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SchemaSerializationException;
-import org.apache.pulsar.client.api.schema.GenericObject;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.api.schema.SchemaInfoProvider;
 import org.apache.pulsar.client.impl.schema.generic.GenericProtobufNativeSchema;
@@ -31,7 +30,10 @@
 import org.apache.pulsar.common.schema.SchemaType;
 
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static com.google.common.base.Preconditions.checkState;
 
@@ -78,27 +80,7 @@ public boolean supportSchemaVersioning() {
 
     @Override
     public GenericRecord decode(byte[] bytes, byte[] schemaVersion) {
-        if (schema == null) {
-            SchemaInfo schemaInfo = null;
-            try {
-                schemaInfo = schemaInfoProvider.getLatestSchema().get();
-                if (schemaInfo == null) {
-                    // schemaless topic
-                    schemaInfo = BytesSchema.of().getSchemaInfo();
-                }
-            } catch (InterruptedException | ExecutionException e ) {
-                if (e instanceof InterruptedException) {
-                    Thread.currentThread().interrupt();
-                }
-                log.error("Can't get last schema for topic {} use AutoConsumeSchema", topicName);
-                throw new SchemaSerializationException(e.getCause());
-            }
-            // schemaInfo null means that there is no schema attached to the topic.
-            schema = generateSchema(schemaInfo);
-            schema.setSchemaInfoProvider(schemaInfoProvider);
-            log.info("Configure {} schema for topic {} : {}",
-                    componentName, topicName, schemaInfo.getSchemaDefinition());
-        }
+        fetchSchemaIfNeeded();
         ensureSchemaInitialized();
         return adapt(schema.decode(bytes, schemaVersion), schemaVersion);
     }
@@ -144,7 +126,7 @@ public Optional<Object> getNativeSchema() {
         }
     }
 
-    private Schema<?> generateSchema(SchemaInfo schemaInfo) {
+    private static Schema<?> generateSchema(SchemaInfo schemaInfo) {
         // when using `AutoConsumeSchema`, we use the schema associated with the messages as schema reader
         // to decode the messages.
         final boolean useProvidedSchemaAsReaderSchema = false;
@@ -247,4 +229,47 @@ public static GenericRecord wrapPrimitiveObject(Object value, SchemaType type, b
     public Schema<?> getInternalSchema() {
         return schema;
     }
+
+    /**
+     * It may happen that the schema is not loaded but we need it, for instance in order to call getSchemaInfo()
+     * We cannot call this method in getSchemaInfo, because getSchemaInfo is called in many
+     * places and we will introduce lots of deadlocks.
+     */
+    public void fetchSchemaIfNeeded() throws SchemaSerializationException {
+        if (schema == null) {
+            if (schemaInfoProvider == null) {
+                throw new SchemaSerializationException("Can't get accurate schema information for topic " + topicName +
+                                                "using AutoConsumeSchema because SchemaInfoProvider is not set yet");
+            } else {
+                SchemaInfo schemaInfo = null;
+                try {
+                    schemaInfo = schemaInfoProvider.getLatestSchema().get();
+                    if (schemaInfo == null) {
+                        // schemaless topic
+                        schemaInfo = BytesSchema.of().getSchemaInfo();
+                    }
+                } catch (InterruptedException | ExecutionException e) {
+                    if (e instanceof InterruptedException) {
+                        Thread.currentThread().interrupt();
+                    }
+                    log.error("Can't get last schema for topic {} using AutoConsumeSchema", topicName);
+                    throw new SchemaSerializationException(e.getCause());
+                }
+                // schemaInfo null means that there is no schema attached to the topic.
+                schema = generateSchema(schemaInfo);
+                schema.setSchemaInfoProvider(schemaInfoProvider);
+                log.info("Configure {} schema for topic {} : {}",
+                        componentName, topicName, schemaInfo.getSchemaDefinition());
+            }
+        }
+    }
+
+    @Override
+    public String toString() {
+        if (schema != null && schema.getSchemaInfo() != null) {
+            return "AUTO_CONSUME(schematype=" + schema.getSchemaInfo().getType() + ")";
+        } else {
+            return "AUTO_CONSUME(uninitialized)";
+        }
+    }
 }

tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/io/TestGenericObjectSink.java

@@ -65,6 +65,8 @@ public void write(Record<GenericObject> record) {
         log.info("value {}", record.getValue());
         log.info("value schema type {}", record.getValue().getSchemaType());
         log.info("value native object {}", record.getValue().getNativeObject());
+
+        record.ack();
     }
 
     @Override

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/PulsarGenericObjectSinkTest.java

@@ -31,6 +31,7 @@
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.policies.data.SinkStatus;
 import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.tests.integration.docker.ContainerExecException;
 import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
 import org.apache.pulsar.tests.integration.suites.PulsarStandaloneTestSuite;
@@ -89,6 +90,9 @@ public void testGenericObjectSink() throws Exception {
                 .serviceUrl(container.getPlainTextServiceUrl())
                 .build();
 
+        @Cleanup
+        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(container.getHttpServiceUrl()).build();
+
         // we are not using a parametrized test in order to save resources
         // we create N sinks, send the records and verify each sink
         // sinks execution happens in parallel
@@ -97,67 +101,61 @@ public void testGenericObjectSink() throws Exception {
                 new SinkSpec("test-kv-sink-input-avro-" + randomName(8), "test-kv-sink-avro-" + randomName(8), Schema.AVRO(Pojo.class), Pojo.builder().field1("a").field2(2).build()),
                 new SinkSpec("test-kv-sink-input-kv-string-int-" + randomName(8), "test-kv-sink-input-kv-string-int-" + randomName(8),
                         Schema.KeyValue(Schema.STRING, Schema.INT32), new KeyValue<>("foo", 123)),
-                new SinkSpec("test-kv-sink-input-kv-avro-json-" + randomName(8), "test-kv-sink-input-kv-string-int-" + randomName(8),
-                        Schema.KeyValue(Schema.AVRO(PojoKey.class), Schema.JSON(Pojo.class)), new KeyValue<>(PojoKey.builder().field1("a").build(), Pojo.builder().field1("a").field2(2).build()))
+                new SinkSpec("test-kv-sink-input-kv-avro-json-inl-" + randomName(8), "test-kv-sink-input-kv-string-int-" + randomName(8),
+                        Schema.KeyValue(Schema.AVRO(PojoKey.class), Schema.JSON(Pojo.class), KeyValueEncodingType.INLINE), new KeyValue<>(PojoKey.builder().field1("a").build(), Pojo.builder().field1("a").field2(2).build())),
+                new SinkSpec("test-kv-sink-input-kv-avro-json-sep-" + randomName(8), "test-kv-sink-input-kv-string-int-" + randomName(8),
+                        Schema.KeyValue(Schema.AVRO(PojoKey.class), Schema.JSON(Pojo.class), KeyValueEncodingType.SEPARATED), new KeyValue<>(PojoKey.builder().field1("a").build(), Pojo.builder().field1("a").field2(2).build()))
         );
-        // submit all sinks
+
+        final int numRecords = 2;
+
         for (SinkSpec spec : specs) {
             submitSinkConnector(spec.sinkName, spec.outputTopicName, "org.apache.pulsar.tests.integration.io.TestGenericObjectSink", JAVAJAR);
-        }
-        // check all sinks
-        for (SinkSpec spec : specs) {
+
             // get sink info
             getSinkInfoSuccess(spec.sinkName);
             getSinkStatus(spec.sinkName);
-        }
 
-        final int numRecords = 10;
-
-        for (SinkSpec spec : specs) {
             @Cleanup Producer<Object> producer = client.newProducer(spec.schema)
                     .topic(spec.outputTopicName)
                     .create();
             for (int i = 0; i < numRecords; i++) {
                 MessageId messageId = producer.newMessage()
                         .value(spec.testValue)
                         .property("expectedType", spec.schema.getSchemaInfo().getType().toString())
+                        .property("recordNumber", i + "")
                         .send();
                 log.info("sent message {} {}  with ID {}", spec.testValue, spec.schema.getSchemaInfo().getType().toString(), messageId);
             }
-        }
 
-        // wait that all sinks processed all records without errors
-        try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(container.getHttpServiceUrl()).build()) {
-            for (SinkSpec spec : specs) {
-                try {
-                    log.info("waiting for sink {}", spec.sinkName);
-                    for (int i = 0; i < 120; i++) {
-                        SinkStatus status = admin.sinks().getSinkStatus("public", "default", spec.sinkName);
-                        log.info("sink {} status {}", spec.sinkName, status);
-                        assertEquals(status.getInstances().size(), 1);
-                        SinkStatus.SinkInstanceStatus instance = status.getInstances().get(0);
-                        if (instance.getStatus().numWrittenToSink >= numRecords) {
-                            break;
-                        }
-                        assertTrue(instance.getStatus().numRestarts > 1, "Sink was restarted, probably an error occurred");
-                        Thread.sleep(1000);
-                    }
 
+            // wait that all sinks processed all records without errors
+
+            try {
+                log.info("waiting for sink {}", spec.sinkName);
+                for (int i = 0; i < 120; i++) {
                     SinkStatus status = admin.sinks().getSinkStatus("public", "default", spec.sinkName);
                     log.info("sink {} status {}", spec.sinkName, status);
                     assertEquals(status.getInstances().size(), 1);
-                    assertTrue(status.getInstances().get(0).getStatus().numWrittenToSink >= numRecords);
-                    assertTrue(status.getInstances().get(0).getStatus().numSinkExceptions == 0);
-                    assertTrue(status.getInstances().get(0).getStatus().numSystemExceptions == 0);
-                    log.info("sink {} is okay", spec.sinkName);
-                } finally {
-                    dumpSinkLogs(spec);
+                    SinkStatus.SinkInstanceStatus instance = status.getInstances().get(0);
+                    if (instance.getStatus().numWrittenToSink >= numRecords) {
+                        break;
+                    }
+                    assertTrue(instance.getStatus().numRestarts > 1, "Sink was restarted, probably an error occurred");
+                    Thread.sleep(1000);
                 }
-            }
-        }
 
+                SinkStatus status = admin.sinks().getSinkStatus("public", "default", spec.sinkName);
+                log.info("sink {} status {}", spec.sinkName, status);
+                assertEquals(status.getInstances().size(), 1);
+                assertTrue(status.getInstances().get(0).getStatus().numWrittenToSink >= numRecords);
+                assertTrue(status.getInstances().get(0).getStatus().numSinkExceptions == 0);
+                assertTrue(status.getInstances().get(0).getStatus().numSystemExceptions == 0);
+                log.info("sink {} is okay", spec.sinkName);
+            } finally {
+                dumpSinkLogs(spec);
+            }
 
-        for (SinkSpec spec : specs) {
             deleteSink(spec.sinkName);
             getSinkInfoNotFound(spec.sinkName);
         }