Hi everyone! Is there an example/receipt to use an external store to store the Flink state? We have a use case that the TTL for the state needs to be configured dynamically for each key but found this is not as apparent so thinking use an workaround.

Why do you need to set the TTL?

Because we have a very large state and it continues growing but some keys became inactive and no longer needed.

Adding a new state backend to Flink is a major undertaking. The backends aren't pluggable. What you could do it is use an existing state backend in combination with keyed timers. That would give you the flexibility you desire, at the expense of storing the timers and having to manage state retention/expiration in your code.

I didn’t think about timers… I think that’s exactly what we need! Thank you David.

If you hold keyed state in a KeyedBroadcastProcessFunction, then you can trigger a clean-up of keyed state with a broadcast message. Or you could take a savepoint, run a batch job that uses the state processor API to clear our unwanted state, and then restart from the now smaller savepoint.

I’m using

KeyedProcessFunction

, can I clear the keyed state using the

clear()

by consuming a message from another stream (non-broadcast)?

With a

KeyedProcessFunction

you can clear the the keyed state for the key currently in context (the key for the event or timer being processed). In a

KeyedBroadcastProcessFunction

you can iterate over all of the keys and do what you want with each piece of state.

Gotcha! Thanks, David!