Hey all, I am trying to implement an enriched DataStream, I would like to enrich a datastream ,which is the joined output of two streams, with another stream that has relatively few records. Basically, I would like to store all of the records from the enrichment stream in state and do a lookup from the joined stream. The enrichment stream is bounded. I have seen a few examples where lookups are done with external data sources like this Flink Forward Talk Example and this AWS example where this enrichment is done in a

RichFlatMapFunction

, I need some guidance on how to load all the records from the enrichment stream in state. What sort of data structure would be used? How do I end shut off the stream after all the records have been loaded? The bounded enrichment stream is from a Kafka topic. Is there a better approach? Thank you.