pyflink kafka连接器将接收到的json数据反序列化为null

from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaConsumer
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.descriptors import Kafka, Json, Schema

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

kafka_properties = {
    'bootstrap.servers': 'kafka_server:9092',
    'group.id': 'flink_consumer_group',
    'auto.offset.reset': 'latest'
}

kafka_topic = 'your_kafka_topic'

t_env.connect(
    Kafka()
    .version('universal')
    .topic(kafka_topic)
    .properties(kafka_properties)
    .start_from_latest()
    .json_schema(
        '{'
        '  "type": "object",'
        '  "properties": {'
        '    "field1": { "type": "null" },'
        '    "field2": { "type": "string" },'
        '    "field3": { "type": "integer" }'
        '  }'
        '}'
    )
).with_format(
    Json()
    .fail_on_missing_field(True)
    .derive_schema()
).in_append_mode().register_table_source('kafka_source')

kafka_table = t_env.from_path('kafka_source')

result_table = kafka_table.select('field1, field2, field3')
result_table.execute_insert('result_table')