flink-1.13.1 ddl kafka????JSON???? (ObjectNode) jsonNode????

[kcz]

????:1.13.1 ????????????:
Caused by: java.lang.ClassCastException: 
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode 
cannot be cast to 
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode
        at 
org.apache.flink.formats.json.JsonToRowDataConverters.lambda$createRowConverter$ef66fe9a$1(JsonToRowDataConverters.java:344)
        at 
org.apache.flink.formats.json.JsonToRowDataConverters.lambda$wrapIntoNullableConverter$de0b9253$1(JsonToRowDataConverters.java:376)
        at 
org.apache.flink.formats.json.JsonRowDataDeserializationSchema.convertToRowData(JsonRowDataDeserializationSchema.java:121)
        at 
org.apache.flink.formats.json.JsonRowDataDeserializationSchema.deserialize(JsonRowDataDeserializationSchema.java:106)



DDL????????:
CREATE TABLE user_behavior (
    user_id string 
) WITH (
 'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json'
);
select * from user_behavior;

????????????:
        StreamExecutionEnvironment streamEnv = 
StreamExecutionEnvironment.createLocalEnvironment(new Configuration());
        StreamTableEnvironment tableEnv = 
StreamTableEnvironment.create(streamEnv);
        tableEnv.executeSql(sql).print();

     pom????????:


      &nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-table-api-java</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-table-planner-blink_${scala.version}</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-connector-hive_${scala.version}</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;


&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-table-api-java-bridge_${scala.version}</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;


&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-table-planner_${scala.version}</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;


&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-clients_${scala.version}</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;


&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-connector-kafka_2.11</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp;&nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; <dependency&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<groupId&gt;org.apache.flink</groupId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<artifactId&gt;flink-json</artifactId&gt;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
<version&gt;${flink.version}</version&gt;
&nbsp; &nbsp; &nbsp; &nbsp; </dependency&gt;