Avro or KeyValue Schema throws exception in PulsarRecordCursor

[eaba]

Describe the bug
Avro or KeyValue Schema throws exception in PulsarRecordCursor

java.lang.NullPointerException: undefined
at org.apache.pulsar.sql.presto.PulsarRecordCursor.advanceNextPosition(PulsarRecordCursor.java:493)
at io.prestosql.spi.connector.RecordPageSource.getNextPage(RecordPageSource.java:90)
at io.prestosql.operator.TableScanOperator.getOutput(TableScanOperator.java:302)
at io.prestosql.operator.Driver.processInternal(Driver.java:379)
at io.prestosql.operator.Driver.lambda$processFor$8(Driver.java:283)
at io.prestosql.operator.Driver.tryWithLock(Driver.java:675)
at io.prestosql.operator.Driver.processFor(Driver.java:276)
at io.prestosql.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:1075)
at io.prestosql.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:163)
at io.prestosql.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:484)
at io.prestosql.$gen.Presto_332__testversion____20210805_035216_2.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)

[gaoran10]

Could you provide more details? Such as how to reproduce this issue, which Pulsar version.

I'll try to use Avro and KeyValue schema to reproduce this issue.

[eaba]

The unit test here could help you to reproduce the error if you can give it a try https://github.com/eaba/SharpPulsar/blob/dev/Tests/SharpPulsar.Test.SQL/SqlTests.cs

[eaba]

Could you provide more details? Such as how to reproduce this issue, which Pulsar version.

I'll try to use Avro and KeyValue schema to reproduce this issue.

Pulsar 2.8.0

[gaoran10]

I think the root cause is the message metadata is the lake of schemaVersion property. You could change the message serialization method to add schema version in message metadata.

[eaba]

@gaoran10 thanks. Null schemaversion was the problem. I have it fixed now.

Thanks a lot!!

[gaoran10]

Maybe you could refer to Java client logic, the producer will register schema before producing messages to make sure the schema saved in BookKeeper, the register schema command will return a schema version, refer to https://github.com/apache/pulsar/blob/v2.8.0/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L609. There is an especial logic that the bytes schema will not be registered, refer to https://github.com/apache/pulsar/blob/v2.8.0/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L575. The command is GET_OR_CREATE_SCHEMA.

[eolivelli]

thank you @gaoran10 for looking into this !