对于Flink CDC，谁有java 开发的基础代码？

import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import io.debezium.data.Envelope;
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.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class FlinkCDCExample {

public static void main(String[] args) throws Exception {

    // Create the execution environment
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // Set up the Kafka consumer properties
    Properties consumerProperties = new Properties();
    consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink-cdc-example");

    // Create the Debezium source function
    DebeziumSourceFunction<Envelope> sourceFunction = DebeziumSourceFunction.builder()
            .hostname("localhost")
            .port(8083)
            .username("debezium")
            .password("dbz-secret")
            .database("inventory")
            .tableList("products")
            .deserializationSchema(new DebeziumDeserializationSchema())
            .build();

    // Create the Kafka consumer
    FlinkKafkaConsumer<Envelope> consumer = new FlinkKafkaConsumer<>("inventory-changes", new SimpleStringSchema(), consumerProperties);

    // Add the Debezium source function to the execution environment
    DataStream<Envelope> debeziumSource = env.addSource(sourceFunction);

    // Add the Kafka consumer to the execution environment
    DataStream<String> kafkaSource = env.addSource(consumer);

    // Print the Debezium source data to the console
    debeziumSource.print();

    // Print the Kafka source data to the console
    kafkaSource.print();

    // Execute the streaming job
    env.execute("Flink CDC Example");
}

}