Savepoint state keeps growing

[Le Xu <sh...@gmail.com>]

Hello!

I'm trying to do a savepoint and reconfigure my window aggregation jobs
overtime and I noticed that the windowed results keep accumulating with
time (I'm using heap backend). I modified the snapshot code and noticed
that Flink does not drop my aggregated window states after the window has
closed -- I'm seeing accumulated window data that looks like

TimeWindow{start=1679953590000, end=1679953600000}
TimeWindow{start=1679953660000, end=1679953670000}
TimeWindow{start=1679953570000, end=1679953580000}...
.....

after modified snapshot logic for heap memory backend. This prevents me to
do savepoints overtime due to the state size and I'm wondering is there a
way for me to force Flink to remove the windowed results after triggered.
I'm testing an older version (1.4.1) and I can't migrate to newer version
due to the changes to the source code. Any suggestions are appreciated.

Thanks,

Le