Kafka connector: read on schema-registry table with schema references failing

[vpnvishv]

When I tried to do a select on a topic/subject that has references to other schemas, getting this error,

[trino@presto-trino-coordinator-564cc6587c-fs5qp /]$ trino --catalog=kafka --schema=default --debug
trino:default> select timestampmsecs from dummy;
Query 20210216_061605_00007_e5knx, FAILED, 2 nodes
http://localhost:8080/ui/query.html?20210216_061605_00007_e5knx
Splits: 17 total, 0 done (0.00%)
CPU Time: 0.0s total,     0 rows/s,     0B/s, 12% active
Per Node: 0.0 parallelism,     0 rows/s,     0B/s
Parallelism: 0.0
Peak Memory: 0B
0.37 [0 rows, 0B] [0 rows/s, 0B/s]
Query 20210216_061605_00007_e5knx failed: "Identifier" is not a defined name. The type of the "identifier" field must be a defined name or a {"type": ...} expression.
org.apache.avro.SchemaParseException: "Identifier" is not a defined name. The type of the "identifier" field must be a defined name or a {"type": ...} expression.
	at org.apache.avro.Schema.parse(Schema.java:1635)
	at org.apache.avro.Schema$Parser.parse(Schema.java:1394)
	at org.apache.avro.Schema$Parser.parse(Schema.java:1382)
	at io.trino.decoder.avro.AvroRowDecoderFactory.create(AvroRowDecoderFactory.java:58)
	at io.trino.decoder.DispatchingRowDecoderFactory.create(DispatchingRowDecoderFactory.java:38)
	at io.trino.plugin.kafka.KafkaRecordSetProvider.getRecordSet(KafkaRecordSetProvider.java:68)
	at io.trino.plugin.base.classloader.ClassLoaderSafeConnectorRecordSetProvider.getRecordSet(ClassLoaderSafeConnectorRecordSetProvider.java:49)
	at io.trino.split.RecordPageSourceProvider.createPageSource(RecordPageSourceProvider.java:50)
	at io.trino.spi.connector.ConnectorPageSourceProvider.createPageSource(ConnectorPageSourceProvider.java:68)
	at io.trino.split.PageSourceManager.createPageSource(PageSourceManager.java:64)
	at io.trino.operator.TableScanOperator.getOutput(TableScanOperator.java:298)
	at io.trino.operator.Driver.processInternal(Driver.java:387)
	at io.trino.operator.Driver.lambda$processFor$9(Driver.java:291)
	at io.trino.operator.Driver.tryWithLock(Driver.java:683)
	at io.trino.operator.Driver.processFor(Driver.java:284)
	at io.trino.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:1076)
	at io.trino.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:163)
	at io.trino.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:484)
	at io.trino.$gen.Trino_352____20210216_034829_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:834)

Here Identifier is a different schema/subject in the schema-registry.