Flink CDC里我一执行滚动开窗，就报下面的错误怎么解决？

Flink CDC里就是消费kafka中的debezium-json格式数据，使用滚动窗口统计1分钟新增的数据量，这个简单的需求，被卡住了。之前一直写的datastream，现在想用flink sql来实现我本来就是消费的kafka数据，为什么还要推到kafka里面？我现在数据源是kafka，kafka中的数据是flink cdc同步到kafka中的，数据有更新，有插入。
// 源表
tableEnv.executeSql("CREATE TABLE KafkaTable (\n" +
" UNSEND_ID STRING,\n" +
" SEND_TIME BIGINT,\n" +
" SENDER_CODE STRING,\n" +
" STATUS STRING,\n" +
" CREATE_TIME BIGINT,\n" +
" ts AS TO_TIMESTAMP_LTZ(CREATE_TIME, 3),\n" +
" WATERMARK FOR ts AS ts - INTERVAL '2' SECOND\n" +
") WITH (\n" +
" 'connector' = 'kafka',\n" +
" 'topic' = 'T_E10_MESSAGE_UNSEND_INFO',\n" +
" 'properties.bootstrap.servers' = 'hadoop01:9092',\n" +
" 'properties.group.id' = 'test',\n" +
" 'scan.startup.mode' = 'latest-offset',\n" +
" 'format' = 'debezium-json'\n" +
")\n");// 开窗聚合
tableEnv.executeSql("select\n" +
" TUMBLE_START(ts, interval '1' MINUTE),\n" +
" TUMBLE_END(ts, interval '1' MINUTE),\n" +
" count(1) as cnt\n" +
"from KafkaTable\n" +
"group by TUMBLE(ts, interval '1' MINUTE)").print();
就是上面的数据，我一执行滚动开窗，就报下面的错误怎么解决？Exception in thread "main" org.apache.flink.table.api.TableException: StreamPhysicalGroupWindowAggregate doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, KafkaTable, watermark=[-(TO_TIMESTAMP_LTZ($4, 3), 2000:INTERVAL SECOND)]]], fields=[UNSEND_ID, SEND_TIME, SENDER_CODE, STATUS, CREATE_TIME])有点理解了，但是为什么我指定'format' = 'debezium-json'就不行呢？