.keyBy(new TwoPhaseSelector())
.process(new LogKeyedProcessFunction())

public class LogKeyedProcessFunction extends KeyedProcessFunction<Tuple3<String,String,String>, CollectorGroup<SlowSqlBody>, CollectorGroup<SlowSqlBody>> {

    private final static Logger logger = LoggerFactory.getLogger(LogKeyedProcessFunction.class);
    private final int phaseWait;
    private volatile ListState<CollectorGroup<SlowSqlBody>> listState;

    
    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<CollectorGroup<SlowSqlBody>> slowSqlCollectorGroupState = new ListStateDescriptor<>("collect-group-list", TypeInformation.of(new TypeHint<CollectorGroup<SlowSqlBody>>() {}));
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.seconds(120))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
		slowSqlCollectorGroupState.enableTimeToLive(ttlConfig);
		listState = getRuntimeContext().getListState(slowSqlCollectorGroupState);
    }

    @Override
    public void processElement(CollectorGroup<SlowSqlBody> value, KeyedProcessFunction<Tuple3<String,String,String>, CollectorGroup<SlowSqlBody>, CollectorGroup<SlowSqlBody>>.Context ctx, Collector<CollectorGroup<SlowSqlBody>> out) throws Exception {
        listState.add(value);
        ctx.timerService().registerProcessingTimeTimer(value.getTimestamp()+phaseWait);
    }

    @Override
    public void onTimer(long timestamp, KeyedProcessFunction<Tuple3<String,String,String>, CollectorGroup<SlowSqlBody>, CollectorGroup<SlowSqlBody>>.OnTimerContext ctx, Collector<CollectorGroup<SlowSqlBody>> out) throws Exception {
        try {
            Tuple3<String,String,String> currentKey = ctx.getCurrentKey();
            Iterable<CollectorGroup<SlowSqlBody>> collectorGroups = listState.get();
            CollectorGroup<SlowSqlBody> collectorGroup = null;
            int index=0;
            for (CollectorGroup<SlowSqlBody> group : collectorGroups) {
                if(collectorGroup == null){
                    collectorGroup = group;
                }else{
                    collectorGroup = collectorGroup.merge(group);
                }
                index++;
            }
            if(collectorGroup!=null){
                out.collect(collectorGroup);
            }
        }finally {
            listState.clear();
        }
    }
}

注册的定时任务是2分钟执行一次
现在发现 LogKeyedProcessFunction 这个算子的checkpoint data size大小 从几kb 持续慢慢涨到几十Mb , 感觉有点不正常。状态做了清空，为何还一直涨，没有停下来的意思