go client producer + schema breaks Pulsar SQL

#### Expected behavior

When using go client to send messages with schema, I expect to be able to use Pulsar SQL to query the data.

#### Actual behavior

tl;dr sql-worker is throwing an exception

I copy&pasted code from `pulsar/schema_test.go` to create a test client that is producing a message with JSON schema and consuming it afterwards. I verified that the go client is able to consume the message and retrieve the correct values by calling `msg.GetSchemaValue()`. But when trying to query the topic using SQL command line, I get an internal error:
```
presto> select * from pulsar."public/default"."goJson";

Query 20210621_103409_00001_r9xrk, FAILED, 1 node
Splits: 18 total, 1 done (5.56%)
0:01 [0 rows, 0B] [0 rows/s, 0B/s]

Query 20210621_103409_00001_r9xrk failed: Internal error

presto>
```

At the same time the sql-worker engine reports an error and throws an exception:
```
2021-06-21T10:34:10.534Z ERROR SplitRunner-2-106 io.prestosql.execution.executor.TaskExecutor Error processing Split 20210621_103409_00001_r9xrk.1.0-0 PulsarSplit{splitId=0, connectorId='pulsar', originSchemaName='goJson', schemaName='public/default', tableName='goJson', splitSize=1, schema='{"type":"record","name":"Example","namespace":"test","fields":[{"name":"ID","type":"int"},{"name":"Name","type":"string"}]}', schemaType=JSON, startPositionEntryId=0, endPositionEntryId=1, startPositionLedgerId=285, endPositionLedgerId=285, schemaInfoProperties={"pulsar":"hello"}} (start = 2.54138549436848E8, wall = 490 ms, cpu = 0 ms, wait = 0 ms, calls = 1)
java.lang.NullPointerException
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____20210621_102833_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)
```

I also tried Avro schema instead of JSON schema, but it is causing the same issue.

I wrote a producer in Python, because I was not sure whether this problem is in Pulsar or in pulsar-client-go. And with the python producer, the sql query worked out of the box. That's why I suspect the problem to be in the go client.

#### Steps to reproduce

Start pulsar container first, then start the presto engine by running `./bin/pulsar sql-worker run` and keep the shell open to see the error logs.

Get the `go_pulsar_schema_test.go` here:
https://gist.github.com/r-funke/452fa154aafef36cf71a6ea240b8a930

Run the go client and see that it successfully produces and consumes a message with schema. Run `./bin/pulsar sql` in pulsar container and query the topic that was created by go client by running `select * from pulsar."public/default"."goJson";`. See that it reports the internal error pasted above (you may need to run go client again, if messages were already deleted). In the sql-worker shell you should see the error and exception pasted above.

To verify that it's working when using the python client, run `py_pulsar_schema_producer.py` from the same gist link above. Then query data in sql:
```
presto> select * from pulsar."public/default"."pyJson";
a | b | c | __partition__ | __event_time__ | __publish_time__ | __message_id__ | __sequence_id__ | __producer_name__ | __key__ | __properties__
-------+---+-------+---------------+-------------------------+-------------------------+----------------+-----------------+-------------------+---------+----------------
Hello | 3 | false | -1 | 1970-01-01 00:00:00.000 | 2021-06-21 10:49:00.524 | (312,2,0) | 2 | standalone-3-2 | NULL | {}
Hello | 4 | false | -1 | 1970-01-01 00:00:00.000 | 2021-06-21 10:49:00.527 | (312,3,0) | 3 | standalone-3-2 | NULL | {}
Hello | 1 | false | -1 | 1970-01-01 00:00:00.000 | 2021-06-21 10:49:00.512 | (312,0,0) | 0 | standalone-3-2 | NULL | {}
Hello | 2 | false | -1 | 1970-01-01 00:00:00.000 | 2021-06-21 10:49:00.519 | (312,1,0) | 1 | standalone-3-2 | NULL | {}
(4 rows)

Query 20210621_112301_00000_ta8x6, FINISHED, 1 node
Splits: 18 total, 18 done (100.00%)
0:07 [4 rows, 510B] [0 rows/s, 75B/s]

presto>
```

In the sql-worker shell, you're now seeing an info instead of an error:
```
2021-06-21T11:23:08.443Z INFO 20210621_112301_00000_ta8x6.1.0-0-102 org.apache.pulsar.sql.presto.PulsarRecordCursor Initializing split with parameters: PulsarSplit{splitId=1, connectorId='pulsar', originSchemaName='pyJson', schemaName='public/default', tableName='pyJson', splitSize=2, schema='{
"name": "Example",
"type": "record",
"fields": [
{
"name": "a",
"type": [
"null",
"string"
]
},
{
"name": "b",
"type": [
"null",
"int"
]
},
{
"name": "c",
"type": [
"null",
"boolean"
]
}
]
}', schemaType=JSON, startPositionEntryId=2, endPositionEntryId=4, startPositionLedgerId=312, endPositionLedgerId=312, schemaInfoProperties={}}
```

#### System configuration
**Pulsar version**: 2.8
**pulsar-client-go**: 0.5.0

Start with pulsar/schema_test.go and the linked go_pulsar_schema_test.go reproduction, then run the Pulsar container, sql-worker, and SQL query described in the issue. Compare the Go producer with py_pulsar_schema_producer.py and inspect the reported PulsarRecordCursor.java:493 failure. Done means messages produced with Go JSON or Avro schemas can be queried through Pulsar SQL without the internal error.

Written by the indexing model from the issue text.