ISSUE-12284: Pulsar SQL BufferUnderflowException after upgrading from 2.7.0 to 2.8.1

[sijie]

Original Issue: apache#12284

---

Describe the bug
Following this upgrade guide to upgrade pulsar cluster node by node, from 2.7.0 to 2.8.1.

Presto CLI version 332 is after the upgrade.

Trying to query in Presto CLI from one of the topics,
apache-pulsar/bin/pulsar sql
select * from pulsar."public/default"."test";
returns
Query 20211006_093830_00017_59f2r failed: java.nio.BufferUnderflowException
(See below for full logs)

However, this only affects some topics. Other topics in the same tenant and namespace can be queried just fine.

Also, using pulsar flink connector and pulsar client python api to consume data from the problematic topics are working as expected, no issues.

Full logs

2021-10-06T09:38:30.837Z        ERROR        remote-task-callback-89        io.prestosql.execution.StageStateMachine        Stage 20211006_093830_00017_59f2r.1 failed  
com.google.common.util.concurrent.UncheckedExecutionException: java.nio.BufferUnderflowException  
        at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2051)
        at com.google.common.cache.LocalCache.get(LocalCache.java:3951)
        at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3974)
        at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4935)
        at org.apache.pulsar.sql.presto.PulsarSqlSchemaInfoProvider.getSchemaByVersion(PulsarSqlSchemaInfoProvider.java:76)
        at org.apache.pulsar.sql.presto.PulsarRecordCursor.advanceNextPosition(PulsarRecordCursor.java:485)
        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____20211005_104810_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.nio.BufferUnderflowException
        at java.nio.Buffer.nextGetIndex(Buffer.java:510)
        at java.nio.HeapByteBuffer.getLong(HeapByteBuffer.java:427)
        at org.apache.pulsar.sql.presto.PulsarSqlSchemaInfoProvider.loadSchema(PulsarSqlSchemaInfoProvider.java:106)
        at org.apache.pulsar.sql.presto.PulsarSqlSchemaInfoProvider.access$000(PulsarSqlSchemaInfoProvider.java:49)
        at org.apache.pulsar.sql.presto.PulsarSqlSchemaInfoProvider$1.load(PulsarSqlSchemaInfoProvider.java:61)
        at org.apache.pulsar.sql.presto.PulsarSqlSchemaInfoProvider$1.load(PulsarSqlSchemaInfoProvider.java:58)
        at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529)
        at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278)
        at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155)
        at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045)
        ... 18 more

PS: Saw similar issue, but that one is about byte schema data, mine is string schema data.

Suspect presto has messed up its cache somehow. Any pointer on how to identify the root cause and/or fix this issue will be greatly appreciated. Thanks!