@Test
public void sqlJobWatermarkTest() {

    List<JSONObject> sourceData = Stream.of(
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:01"),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:02"),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:00:03"),
            new JSONObject().fluentPut("data_time", "2020-01-01 12:01:01")
    ).collect(Collectors.toList());
    String sql = "create table t1( " +
            "data_time STRING, " +
            "row1_time AS to_timestamp(data_time)," +
            "WATERMARK FOR row1_time AS row1_time - INTERVAL '5' SECOND " +
            ") with ( 'connector' = 'collection','data'='" + JSON.toJSONString(sourceData) + "');" +
            "create table t2(stime TIMESTAMP(3),cnt bigint) with ( 'connector' = 'print');" +
            "insert into t2 select TUMBLE_START(row1_time, INTERVAL '1' MINUTE) as stime,count(1) cnt from t1 group by TUMBLE(row1_time, INTERVAL '1' MINUTE);;";
    Exception exception = null;
    try {
        SqlConfig config = SqlConfig.builder().sql(sql).jobName("sql_job_test").build();
        SqlJob sqlJob = new SqlJob(config);
        sqlJob.start();
    } catch (Exception e) {
        log.error("sqlJobTest error", e);
        exception = e;
    }
    assert exception == null;
}