flink 注册kafka得csv格式

用flink注册kafka得消息源，format选择csv报错,选择avro可以。是不是kafka连接器不支持csv格式。
报错如下：Exception in thread "main" org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.DeserializationSchemaFactory' in
the classpath。

示例代码如下：

tableEnv.connect(
  new Kafka()
    .version("0.11")
      .topic("result_count")
    .property("bootstrap.servers", "**")
)
  .withFormat(
    new Csv().fieldDelimiter(",")
  )
  .withSchema(
    new Schema()
      .field("world", Types.STRING)
      .field("count", Types.INT)
  )
  .inAppendMode()
  .registerTableSource("result_count")