Flink1.18.0版本连接外部kafka本地执行测试代码报错-[阿里云_云淘科技]

匿名 • 2023年12月29日上午11:31 • 阿里云

版本：Flink1.18.0
代码如下：“`
public static void main(String[] args) {

    //1、创建TableEnvironment
    EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
    TableEnvironment tEnv = TableEnvironment.create(settings);

    //2、创建source  table
    Schema schema = Schema.newBuilder()
            .column("user", DataTypes.STRING())
            .column("url",DataTypes.STRING())
            .column("cTime", DataTypes.STRING())
            .build();
    tEnv.createTemporaryTable( "sourceTable", TableDescriptor.forConnector("kafka")
            .schema(schema)
            .option("topic","spring_test_kafka_topic")
            .option("properties.bootstrap.servers","xxxxx:9092")
            .option("properties.group.id","clicklog")
            .option("scan.startup.mode","earliest-offset")
            .option("format","json")
            .build()
    );

    //3、创建sink  table
    tEnv.createTemporaryTable("sinkTable",
            TableDescriptor.forConnector("print").schema(schema).build());

    //4、执行sql查询
    //Table resultTable = tEnv.sqlQuery(" select ,userurl,cTime from sourceTable ");
    Table resultTable = tEnv.from("sourceTable")
            .select($("user"), $("url"),$("cTime"));

    //5、输出
    resultTable.executeInsert("sinkTable");
}

“`

操作流程：运行代码报错：

请问是什么原因呢，请大佬指点