Github user twalthr commented on the issue:
https://github.com/apache/flink/pull/5327
Thanks for the update @hequn8128. The changes look good. I tested your
implementation on a cluster with TPC-H data. The results were equal to the
batch results and the state clean-up worked. I will merge this :-)---
