@Test
    public void testKafkaWithSqlFile() throws Exception {
        final String topic = "test-topic";
        initializeKafkaTable(topic);

        submitSQLJob(sqlFile, kafkaJar, starrocksJar, mysqlJar, mysqlJdbcJar);
        waitUntilJobRunning(Duration.ofSeconds(10));

        // generate input
        try (Connection conn = DriverManager.getConnection(MYSQL_CONTAINER.getJdbcUrl(),
                MYSQL_CONTAINER.getUsername(), MYSQL_CONTAINER.getPassword());
                Statement stat = conn.createStatement()) {
            stat.execute("INSERT INTO test_input VALUES (1,'jacket','water resistant white wind breaker');");
            stat.execute("INSERT INTO test_input VALUES (2,'scooter','Big 2-wheel scooter ');");
        } catch (SQLException e) {
            LOG.error("Update table for CDC failed.", e);
            throw e;
        }

        JdbcProxy proxy = new JdbcProxy(STAR_ROCKS.getJdbcUrl(), STAR_ROCKS.getUsername(),
                STAR_ROCKS.getPassword(),
                STAR_ROCKS.getDriverClassName());

        List<String> expectResult = Arrays.asList(
                "1,jacket,water resistant white wind breaker",
                "2,scooter,Big 2-wheel scooter ");
        proxy.checkResultWithTimeout(
                expectResult,
                "test_output1",
                3,
                60000L);
        // check log appender
        String logs = OPEN_TELEMETRY_CONTAINER.getLogs();
        if (!logs.contains("OpenTelemetryLogger installed")) {
            throw new Exception("Failure to append logs to OpenTelemetry");
        }
    }
Test flink sql mysql cdc to starrocks.

@throws Exception The exception may throw when execute the case
inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Kafka2StarRocksTest.java