@BeforeClass
    public static void before() {
        LOG.info("Starting containers...");
        jobManager =
                new GenericContainer<>("flink:1.13.5-scala_2.11-java8")
                        .withCommand("jobmanager")
                        .withNetwork(NETWORK)
                        .withNetworkAliases(INTER_CONTAINER_JM_ALIAS)
                        .withExposedPorts(JOB_MANAGER_REST_PORT, DEBUG_PORT)
                        .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
                        .withLogConsumer(new Slf4jLogConsumer(JM_LOG));
        taskManager =
                new GenericContainer<>("flink:1.13.5-scala_2.11-java8")
                        .withCommand("taskmanager")
                        .withNetwork(NETWORK)
                        .withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
                        .withExposedPorts(DEBUG_PORT)
                        .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
                        .dependsOn(jobManager)
                        .withLogConsumer(new Slf4jLogConsumer(TM_LOG));

        Startables.deepStart(Stream.of(jobManager)).join();
        Startables.deepStart(Stream.of(taskManager)).join();
        LOG.info("Containers are started.");
    }
End to end base test environment for test sort-connectors.
Every link : MySQL -> Xxx (Test connector) -> MySQL
inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.13/src/test/java/org/apache/inlong/sort/tests/utils/FlinkContainerTestEnv.java