Better Scaling Experience for KStream

In our day-to-day work, we found it really hard to scale up a stateful stream application when its state store is very heavy. The caveat is that when the newly spinned hosts take ownership of some active tasks, so that they need to use non-trivial amount of time to restore the state store from changelog topic. The reassigned tasks would be available for unpredicted long time, which is not favorable. Secondly the current global rebalance stops the entire application process, which in a rolling host swap scenario would suggest an infinite resource shuffling without actual progress.

Following the community's cooperative rebalancing proposal, we need to build something similar for KStream to better handle the auto scaling experience.