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)
