Github user zhijiangW commented on the issue:
https://github.com/apache/flink/pull/6257
Thanks for fixing this bug, and we also solve this problem in this way.
This solution seems more lightweight than the way in
[6254](https://github.com/apache/flink/pull/6254), and I also think the lock
adjusting in `6254` has reference values.---
