有flink cdc连接tidb的datastream代码？

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkCDCExample {
public static void main(String[] args) throws Exception {
// 设置执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

    // 创建 Kafka 数据源
    String kafkaBootstrapServers = "localhost:9092";
    String kafkaTopic = "my_topic";
    FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(kafkaTopic, new SimpleStringSchema(), kafkaProperties);
    DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

    // 连接 TiDB CDC
    Table cdcTable = tEnv.fromDataStream(kafkaStream, "message");
    tEnv.createTemporaryView("cdc_table", cdcTable);

    // 执行 SQL 查询
    Table resultTable = tEnv.sqlQuery("SELECT message FROM cdc_table WHERE message LIKE '%keyword%'");

    // 输出结果
    DataStream<String> resultStream = tEnv.toAppendStream(resultTable, String.class);
    resultStream.map(new MapFunction<String, String>() {
        @Override
        public String map(String value) throws Exception {
            return "Result: " + value;
        }
    }).print();

    // 执行任务
    env.execute("Flink CDC Example");
}

}