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

#### 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