Github user zentol commented on the issue:
https://github.com/apache/flink/pull/3838
I've been digging into this for the past week. I found a number of things
to improve and did so in a local branch. Once I've finalized/tests things
(probably tomorrow) I'll link the branch here or open another PR.---
