Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5997
This does fix the issue, but I'd like to extend this a bit. The root issue
is that we're registering the handler for an arbitrary header, which is
ridiculously error-prone.
So i propose changing all instances of
`handlers.add(Tuple2.of(<arbitrary_headers>, <handler>));`
to
`handlers.add(Tuple2.of(<handler>.getMessageHeaders(), <handler>));`---
