title: How is keyed state managed for KeyedBroadcastProcessFunction in Flink?
body: <p>I am using <code>BroadcastState</code> to perform streaming computation in Flink. I have defined a class extending <code>KeyedBroadcastProcessFunction</code> for my job. Say I have a stream A which is keyed by <code>(user_id, location)</code>, and a stream B, which is broadcasted to all executors to process elements in A using my defined class. I understand I can registered a timer in <code>processBroadcastElement</code> or <code>processElement</code> in this class so that when it times out, I can delete the associated state for a specific key group by calling <code>state.clear()</code>. I wonder after that, does this key group still exist?</p>
  
  <p>For example, in stream A, a new message comes with <code>(user_id=1, location='usa')</code> and we have such key group and its associated states generated. After that if another message with <code>(user_id=1, location='usa')</code> comes, it will trigger <code>processElement()</code> and emit result. </p>
  
  <p>Say after 24 hours, I'm no longer interested with this key group <code>(user_id=1, location='usa')</code>, I can register a timer to clear the associated state, but I have no control of this key group. As a result, after 24 hours, when another message with <code>(user_id=1, location='usa')</code> comes, since this key group still exists, <code>processElement()</code> will still be invoked. As the job runs, although their associated states will be cleared after 24 hours, will key groups accumulate or that should not be a concern for memory usage?</p>
  
  <p>Relevant blogs: <a href="https://www.da-platform.com/blog/a-practical-guide-to-broadcast-state-in-apache-flink" rel="nofollow noreferrer">https://www.da-platform.com/blog/a-practical-guide-to-broadcast-state-in-apache-flink</a></p>