Pulsar sql EOF exception when using go client to send empty payload

[longtengz]

Expected behavior

When sending empty payload through go client, pulsar sql should delete the rows specified by the message key.

Actual behavior

I tried go client of version 0.5.0, 0.6.0, 0.7.0 with pulsar of version 2.7.2, 2.7.3, 2.8.0. And all of them had the same EOF exception.

org.apache.pulsar.client.api.SchemaSerializationException: java.io.EOFException
        at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:106)
        at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:42)
        at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.read(AbstractMultiVersionReader.java:67)
        at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.read(AbstractMultiVersionReader.java:73)
        at org.apache.pulsar.client.impl.schema.AbstractStructSchema.decode(AbstractStructSchema.java:70)
        at org.apache.pulsar.sql.presto.AvroSchemaHandler.deserialize(AvroSchemaHandler.java:69)
        at org.apache.pulsar.sql.presto.PulsarRecordCursor.advanceNextPosition(PulsarRecordCursor.java:431)
        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____20211206_071408_2.run(Unknown Source)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.EOFException
        at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:509)
        at org.apache.avro.io.BinaryDecoder.readInt(BinaryDecoder.java:149)
        at org.apache.avro.io.ValidatingDecoder.readInt(ValidatingDecoder.java:83)
        at org.apache.avro.generic.GenericDatumReader.readInt(GenericDatumReader.java:549)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:186)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
        at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:248)
        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:237)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:170)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:144)
        at org.apache.pulsar.client.impl.schema.generic.GenericAvroReader.read(GenericAvroReader.java:101)
        ... 19 more

Steps to reproduce

1. create topic
2. send a few messages with message key
3. use pulsar sql to select from that topic
4. returns correctly
5. send an empty payload with one of the message keys
6. use pulsar sql to select again
7. EOF exception

Note, enable bookkeeperExplicitLacIntervalInMills to see the latest results in pulsar sql immediately.

System configuration

Pulsar version: x.y

[wolfstudy]

@longtengz This looks more like a Pulsar SQL bug?

[longtengz]

The thing is I used go client to produce messages in any shape or form (e.g. only nil out one nullable field), and none of them came back successfully in pulsar sql. But the ones sent by python client works as expected. This issue https://github.com/apache/pulsar/issues/12465 might be related.