@Test
    public void testKafkaWithGroupFile() throws Exception {
        final String topic = "test_topic_for_group_file";
        final String mysqlInputTable = "test_input_for_group_file";
        final String mysqlOutputTable = "test_output_for_group_file";
        initializeMysqlTable("kafka_test_mysql_init.txt", new HashMap() {

            {
                put("MYSQL_INPUT_TABLE", mysqlInputTable);
                put("MYSQL_OUTPUT_TABLE", mysqlOutputTable);
            }
        });
        initializeKafkaTable("kafka_test_kafka_init.txt", new HashMap() {

            {
                put("TOPIC", topic);
                put("ZOOKEEPER_PORT", KafkaContainer.ZOOKEEPER_PORT);
            }
        });
        String groupFile = getGroupFile("kafka_test.json", new HashMap() {

            {
                put("MYSQL_INPUT_TABLE", mysqlInputTable);
                put("MYSQL_OUTPUT_TABLE", mysqlOutputTable);
                put("TOPIC", topic);
                put("ZOOKEEPER_PORT", KafkaContainer.ZOOKEEPER_PORT);
            }
        }).toString();
        submitGroupFileJob(groupFile, kafkaJar, jdbcJar, mysqlJar, mysqlJdbcJar);
        waitUntilJobRunning(Duration.ofSeconds(30));

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

        // validate output
        JdbcProxy proxy =
                new JdbcProxy(MYSQL.getJdbcUrl(), MYSQL.getUsername(), MYSQL.getPassword(), MYSQL_DRIVER_CLASS);
        List<String> expectResult =
                Arrays.asList(
                        "1,jacket,water resistent white wind breaker,0.2,null,null,null",
                        "2,scooter,Big 2-wheel scooter ,5.18,null,null,null");

        proxy.checkResultWithTimeout(
                expectResult,
                mysqlOutputTable,
                7,
                60000L);
    }
Test flink sql mysql cdc to hive

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