Hi all, I’m running into a pretty bizarre issue with a Flink job that leverages Broadcast state of roughly ~1000 records that share 1-2 properties each (not very big). However, I’m finding as the job runs and data is flowing through, the JVM Heap Memory Usage seems to grow unbounded until ultimately failing the job with an OOM as seen below. The actual graph itself uses a broadcast stream to enrich multiple other streams against, something like:

/* Broadcast States (there are two similar to these) */
val broadcastStreamA = streamEnv
            .fromSource(...)
            .flatMap(EnrichmentSourceAMapper())
            .broadcast(BroadcastStateDescriptors.a)

val broadcastStreamB = streamEnv
            .fromSource(...)
            .flatMap(EnrichmentSourceBMapper())
            .broadcast(BroadcastStateDescriptors.b)

/* Other Streams (Enriched By B) */
val stream1 = streamEnv
            .fromSource(...)
            .flatMap()
            .connect(broadcastStreamB)
            .process(BEnricher())

val stream2 = streamEnv
            .fromSource(...)
            .flatMap()
            .connect(broadcastStreamB)
            .process(BEnricher())

/* All Other Streams Enriched By A */
val commonStream = stream1
            .union(stream2, ... streamN)
            .connect(broadcastStreamA)
            .process(AEnricher())

/* Various Calls to Sinks Here */

Does anything seem obviously wrong here? The addition of Broadcast state to the job is a fairly recent add, however the overall size for all of the records consumed in the Broadcast State streams is < 3MB, however the job has over 3GB of RAM and we’re still seeing steady climbs until eventually encountering an OOM exception. The parallelism is only set to 2, so the “explosion” in size has me concerned that the state is being duplicated far more than expected. Any ideas would be greatly appreciated (and obviously happy to share additional details etc.)