如何在Flink中通过TestContainers使用Cassandra接收器

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>cassandra</artifactId>
    <version>1.15.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.13.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.12</artifactId>
    <version>1.13.2</version>
</dependency>

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.cassandra.CassandraSink;
import org.apache.flink.streaming.connectors.cassandra.ClusterBuilder;
import org.testcontainers.containers.CassandraContainer;
import org.testcontainers.utility.DockerImageName;

public class FlinkCassandraTest {

    private static final CassandraContainer<?> cassandra = new CassandraContainer<>(DockerImageName.parse("cassandra:3.11.10"))
            .withInitScript("init.cql"); // 可选：用于初始化数据库的脚本

    @BeforeAll
    public static void setUp() {
        cassandra.start();
    }

    @AfterAll
    public static void tearDown() {
        cassandra.stop();
    }

    @Test
    public void testFlinkJobWithCassandra() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> dataStream = env.fromElements("key1,value1", "key2,value2");

        ClusterBuilder CassandraClusterBuilder = new ClusterBuilder() {
            @Override
            protected Cluster buildCluster(Cluster.Builder builder) {
                return builder.addContactPoint(cassandra.getContainerIpAddress()).build();
            }
        };

        CassandraSink.addSink(dataStream)
                .setQuery("INSERT INTO mykeyspace.mytable (key, value) VALUES (?, ?);")
                .setClusterBuilder(CassandraClusterBuilder)
                .build();

        env.execute("Flink Cassandra Test Job");
    }
}