[jira] [Created] (KAFKA-21090) Improve in-memory state store re-use during a rebalance.

["Matthias J. Sax (Jira)" <ji...@apache.org>]

Matthias J. Sax created KAFKA-21090:
---------------------------------------

             Summary: Improve in-memory state store re-use during a rebalance.
                 Key: KAFKA-21090
                 URL: https://issues.apache.org/jira/browse/KAFKA-21090
             Project: Kafka
          Issue Type: Improvement
          Components: streams
            Reporter: Matthias J. Sax


While working on warm-up task support for "streams" group, in particular the new refiner step, I stumbled upon the issue, that two members (StreamsThread) of the same process/client/instance, cannot hand over tasks to each other directly, but the old owner needs to close the task (dropping in-memory state) before the new owner can take over (and now needs to rebuild the state from the changelog topic). – We only support direct active/standby/warmup task _conversion_ (which is technically a revoke+grant of the same task of a different type) which preserve in-memory state stores.

While we would expect that the task-assignor tries to assign a task to its previous owner (if it already decided to place a task on the same process), it cannot be guaranteed.

We should consider to allow a client to hand over a task from thread-to-thread w/o the need to close it. Ie, the previous owner could "suspend" the task only, preserving the in-memory state store while acking the revocation back to the broker afterwards, allowing the broker to grant the task to the sibling member afterwards, which can get the suspended task from the old owner, and "resume" it.

We would need some protocol changes to add metadata to a "revoke task" instruction to allow a member to distinguish a "full revocation" (what we implement now), vs a "suspend revocation". On a grant, a member would always check if there is a suspended task it can take over, and only create a new one if no suspended task is available.

[Side note: we should also apply some cleanup/timeout on "suspended" task, and close them, to free up the memory, if they are not re-claimed in a timely manner – similar to existing task-cleanup-interval [maybe we could even re-use the exact same config?])

It might also be beneficial, to add persistent vs in-memory state store information to the "topology metadata/description" sent to the broker, allow the refiner (and maybe also the assignor?) to make smarter decisions about task placement, and warm-up task injection etc.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)