大佬们，Flink CDC 中谁有mysql动态获取表结构变更的示例呀，

Flink CDC 中可以通过 MySQL Connector 和 Hive Connector 实现动态获取表结构变更。以下是使用 MySQL Connector 的示例代码：

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerConfig;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaTopicPartitioner;
import org.apache.flink.table import DataTypes, StreamTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironmentImpl;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.KeyedDelimitedTextFileSource;
import org.apache.flink.table.types.logicaltypes.DecimalTypeInfo;
import org.apache.flink.table.types.logicaltypes.TimeTypeInfo;
import org.mysql.jdbc2.optional.MysqlConnectionPool;
import org.mysql.jdbc2.optional.MysqlDataSource;

import java.io.IOException;
import java.util.Properties;

public class MysqlCDCExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置 WatermarkStrategy 为 EventTime,这样可以支持动态获取表结构变更信息
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        // 配置 Kafka 相关连接信息
        Properties props = new Properties();
        props.setProperty("bootstrapServers", "localhost:9092");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), props);
        consumer.setStartFromLatest(); // 从最新数据开始消费
        DataStream<String> inputStream = env
                .addSource(consumer)
                .name("input-stream");
        
        // 配置 MySQL Connector 相关连接信息
        String url = "jdbc:mysql://localhost:3306/test?useSSL=false&serverTimezone=UTC";
        String user = "root";
        String password = "password";
        String tableName = "test_table";
        MysqlDataSource dataSource = new MysqlDataSource();
        dataSource