protected abstract void modifyDataSourceProperties(RoutineLoadDataSourceProperties dataSourceProperties)
            throws DdlException;
@Test
    public void testModifyDataSourceProperties() throws Exception {
        KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();
        ConnectContext connectContext = UtFrameUtils.createDefaultCtx();
        //alter data source custom properties
        String groupId = "group1";
        String clientId = "client1";
        String defaultOffsets = "OFFSET_BEGINNING";
        String originStmt = "alter routine load for db.job1 " +
                "FROM KAFKA (" +
                "   \"property.group.id\" = \"" + groupId + "\"," +
                "   \"property.client.id\" = \"" + clientId + "\"," +
                "   \"property.kafka_default_offsets\" = \"" + defaultOffsets + "\"" +
                ")";
        routineLoadJob.setOrigStmt(new OriginStatement(originStmt, 0));
        AlterRoutineLoadStmt stmt = (AlterRoutineLoadStmt) UtFrameUtils.parseStmtWithNewParser(originStmt, connectContext);
        routineLoadJob.modifyJob(stmt.getRoutineLoadDesc(), stmt.getAnalyzedJobProperties(),
                stmt.getDataSourceProperties(), new OriginStatement(originStmt, 0), true);
        routineLoadJob.convertCustomProperties(true);
        Map<String, String> properties = routineLoadJob.getConvertedCustomProperties();
        Assert.assertEquals(groupId, properties.get("group.id"));
        Assert.assertEquals(clientId, properties.get("client.id"));
        Assert.assertEquals(-2L,
                (long) Deencapsulation.getField(routineLoadJob, "kafkaDefaultOffSet"));
    }