Apache flink 1.52 Rowtime时间戳为空

我正在使用以下代码进行一些查询：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Row> ds = SourceHelp.builder().env(env).consumer010(MyKafka.builder().build().kafkaWithWaterMark2())
        .rowTypeInfo(MyRowType.builder().build().typeInfo())
        .build().source4();
//,proctime.proctime,rowtime.rowtime
String sql1 = "select a,b,max(rowtime)as rowtime from user_device group by a,b";
DataStream<Row> ds2 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device").fields("a,b,rowtime.rowtime")
        .rowTypeInfo(MyRowType.builder().build().typeInfo13())
        .sql(sql1).in(ds).build().result();

ds2.print();
// String sql2 = "select a,count(b) as b from user_device2 group by a";
String sql2 = "select a,count(b) as b,HOP_END(rowtime,INTERVAL '5' SECOND,INTERVAL '30' SECOND) as c from user_device2 group by HOP(rowtime, INTERVAL '5' SECOND, INTERVAL '30' SECOND),a";
DataStream<Row> ds3 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device2").fields("a,b,rowtime.rowtime")
        .rowTypeInfo(MyRowType.builder().build().typeInfo14())
        .sql(sql2).in(ds2).build().result();

ds3.print();
env.execute("test");

注意：对于sql1，我使用带有rowtime的max函数，它不起作用，并抛出以下异常：

线程“main”中的异常org.apache.flink.runtime.client.JobExecutionException：java.lang.RuntimeException：Rowtime时间戳为null。请确保定义了正确的TimestampAssigner，并且流环境使用EventTime时间特性。在org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking（MiniCluster.java:625）在org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute（LocalStreamEnvironment.java:123）在com.aicaigroup.water .WaterTest.testRowtimeWithMoreSqls5（WaterTest.java:158）at com.aicaigroup.water.WaterTest.main（WaterTest.java:20）引起：java.lang.RuntimeException：Rowtime timestamp为null。请确保定义了正确的TimestampAssigner，并且流环境使用EventTime时间特性。

然后我尝试像这样更新sql1“从user_device中选择a，b，rowtime”，它可以工作。那么如何修复错误呢？第一个sql应该使用group by，第二个sql应该使用timeWindow的rowtime。