Hi Team, I am getting below error while apply SQL on table, Can anyone help? Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = keyur_log, partition = 0, leaderEpoch = 1, offset = 400, CreateTime = 1692944973855, serialized key size = -1, serialized value size = 6, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@605493e5). at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57) at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53) Caused by: java.lang.NullPointerException: metadataHandlerProvider at java.base/java.util.Objects.requireNonNull(Objects.java:246) at org.apache.calcite.rel.metadata.RelMetadataQueryBase.getMetadataHandlerProvider(RelMetadataQueryBase.java:122) at org.apache.calcite.rel.metadata.RelMetadataQueryBase.revise(RelMetadataQueryBase.java:118) at org.apache.calcite.rel.metadata.RelMetadataQuery.getNonCumulativeCost(RelMetadataQuery.java:333) at org.apache.calcite.plan.volcano.VolcanoPlanner.getCost(VolcanoPlanner.java:727) at org.apache.calcite.plan.volcano.VolcanoPlanner.getCostOrInfinite(VolcanoPlanner.java:714) at org.apache.calcite.plan.volcano.VolcanoPlanner.propagateCostImprovements(VolcanoPlanner.java:971) at org.apache.calcite.plan.volcano.VolcanoPlanner.addRelToSet(VolcanoPlanner.java:1408) at org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1368) at org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:598) at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRootConverters(VolcanoPlanner.java:571) at org.apache.calcite.plan.volcano.VolcanoPlanner.setRoot(VolcanoPlanner.java:279) at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:316) at org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:62) at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:59) at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156) at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156) at scala.collection.Iterator.foreach(Iterator.scala:937) at scala.collection.Iterator.foreach$(Iterator.scala:937)

It’s a failure to deserialize a record read from Kafka. Something is expecting to have metadata, but it’s missing. How did you define the Kafka connection?

Using Kafka connector

Obviously, but you probably have something there that doesn’t agree with Kafka.

"CREATE TABLE LogTable (\n" +
        "  `f0` STRING\n" +
        ") WITH (\n" +
        "  'connector' = 'kafka',\n" +
        "  'topic' = 'keyur_log',\n" +
        "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
        "  'properties.group.id' = 'testGroup',\n" +
        "  'properties.auto.offset.reset' = 'latest'," +
        "  'value.format' = 'raw'\n" +
        ")"

Here is the connector