Just to follow up on this issue: after collecting some data and setting up additional tests we have managed to pinpoint the issue to the the ScalaBuff-generated code that decodes enumerations. After switching to use ScalaPB generator instead, the problem was gone.
One thing peculiar about this bug, however, is that it seems to manifest only on Flink. We have a number of ad-hoc streaming pipelines (without Flink) that are still using the very same decoder code and have been running for weeks without seemingly experiencing any memory or performance issues. The versions of Flink that we saw this happening this on are 1.0 and 1.0.1.