...
For a clean upgrade path for RocksDB itself, we need to introduce the above configs and also change the user Subscription
and Assigment
rebalance metadata. The metadat itself must not be changed but For this change we bump the version number from 2 to 3 to distinguesh between old and new instances with within a consumer group. New instances will encode a version number for there local stores. Becuase of this rebalance metadata change, we suggest to also fix KAFKA-6054 as mentioned above. Note that version 0.10.0.x used metadata version 1, while versions 0.10.1.x, ..., 1.1.x use metadata version 2.
...
- In-place upgrade: this requires two rolling bounces of each application instance
- advantage: simpler than roll-over upgrade
- disadvantage: need 2x local disk storage during upgrade
- upgrade flow:
- prepare a jar hot swap from old version (not 0.10.0.x) to 1.2; Kafka Streams need to be configured with
upgrade.mode="in_place"
for startup - do a rolling bounce to get the new jar and config in place for each instance
- config
"in_place"
tells the application to sendSubscription
in version 2 (to be compatible with potential version 2 leader in the group) - instances will receive a version 2
Assigmentin
in this stage
- config
- each instance will create a "back groud" task, that start to rebuild RocksDB stores in new format in parallel to existing RocksDB
- existing RocksDB will be used for processing so there is no down time
- each instance builds up the new RocksDB state for all its assigned active and standby tasks
- when all stores are ready, we print a INFO message in the log (thus logs must be watched)
- after all stores are prepared, user prepares a second round of rebalance; this time, the configuration parameter
upgrade.mode
must be removed for new startup - do a second rolling bounce for each instance to get new config
- bounced instance can send a version 3
Subscription
; they encode each store twice (once with storage format 1 and once with storage format 2) - the leader sends version 2
Assigment
(based on reported old stores with format 1) back to indicate that upgrade is not completed yet as long as at least one version 2Subsription
is received - as long as version 2
Assigment
is received, the old store will be used for processing and the new store will be futher updated in the back ground - if the leader receives only version if the leader receives only version 3
Subscirption,
it send version 3Assigment
(based on reported new stores with format 2) back indicating that the upgrad is completed - when Kafka Streams receives a version 3
Subscription
is check for local RocksDB directories in old format 1 and deletes them to free disk space
- bounced instance can send a version 3
- prepare a jar hot swap from old version (not 0.10.0.x) to 1.2; Kafka Streams need to be configured with
- upgrading from 0.10.0.x to 1.2 uses the same upgrage pattern, however config
upgrade.mode="in_place_0.10.0.x"
must be used- instead of old
Subscription
andAssigmentmet
metadata verions 2, metadata version 1 is used
- instead of old
- Roll-over upgrade:
- in-place upgrade might not be feasible because of its large local disk storage requirement; thus, roll-over is an alternative to enable upgrading if not enough local disk space is availalbe for a in-place upgrad
- if an application is running with N instances, user starts N new instances in parallel (the all form one consumer group); the new intances rebuild the RocksDB state with the new format; finally, the old instances are stopped
- upgrade flow:
- prepare a jar hot swap from old version (not 0.10.0.x) to 1.2 for all existing (old) instances; Kafka Streams need to be configured with
upgrade.mode="roll_over"
for startup - do a rolling bounce to get the new jar and config in place for each instance
- config
"roll_over"
tells the application to sendSubscription
in version 2 (tp be compatible with potential version 2 leader in the group) - instances will receive a version 2
Assigmentin
in this stage - all old instances will just resume processing as usual
- config
- users starts N new instances with config parameter
upgrade.mode
removed- the new instances send
Subscription
with version 3 - the leader can distinguesh between old and new instances based on the used
Subscription
version and encoded information- note, for the in-place upgrade case, new instances will report old and new store format for all their stores (as long as upgrade is not completed)
- new roll-over nodes, will only report new format stores (and never both)
- thus, the leader receives mixed version 2 and version 3 meta data, it can figure out if this is a in-place or rolling upgrade (if a 1.2 leader only receives version 2 metadata it cannot distingues between both upgrade modes, but there is also no need to disntinguesh both for this case as the leader behaves exacty the same for both upgrad protocols in this phase)
- the leader sends a version 2
Assignment
to the old instances and "mirrors" the assignment for the new instances- in the "mirror" assignment, aktive tasks are replaced by standby tasks and it's send as version 3
Assignment
- because the leader only assigns standby taks to the new instances, they only start to build up the stores in new storage format
- we need to report the progress of the StandbyTasks and write and INFO log message if they are ready to take over
- in the "mirror" assignment, aktive tasks are replaced by standby tasks and it's send as version 3
- the new instances send
- all old instances are stopped
- as long as at least one old instance is alive, the leader assign all aktive tasks to the old instances only
- however, as those are shut down in a row, idealy a rebalance will only trigger after all are down already (note, Kafka Streams does not send a "leave group request" and thus a rebalance only happens after session.timeout passes)
- when all old instances are down, the leader only receives version 3
Subscriptions
and computes are version 3Assigment
base on the prepared stores and upgrade is finished
- prepare a jar hot swap from old version (not 0.10.0.x) to 1.2 for all existing (old) instances; Kafka Streams need to be configured with
- upgrading from 0.10.0.x to 1.2 uses the same upgrage pattern, however config
upgrade.mode="roll_over_0.10.0.x"
must be used- instead of old
Subscription
andAssigmentmet
metadata verions 2, metadata version 1 is used
- instead of old
- Offline upgrade:
- this is the simplest form of upgrading but results in application downtime
- stop all running instances
- swap the jar and restart all instances (config
upgrade.mode
is not set)- the new instances will send and receive version 3
Subscription
andAssignments
- the will not find any store in new format and just recorver the stores using the new format from the changelog topic
- as the receive version 3
Assignemnt
they delete local RocksDB directories with old format
- the new instances will send and receive version 3
- Upgrading from 0.10.0.x to 0.10.1.x, ..., 1.1.x
- upgrade flow:
- prepare a jar hot swap from 0.10.0.x to 1.2; Kafka Streams need to be configured with
upgrade.mode="0.10.0.x"
for startup - do a rolling bounce to get the new jar and config in place for each instance
- config
"0.10.0.x"
tells the application to sendSubscription
in version 1 (to be compatible with potential version 1 leader in the group) - instances will receive a version 1
Assigmentin
in this stage
- config
- do a second rolling bounce for each instance with config parameter
upgrade.mode
removed- bounced instance can send a version 2
Subscription
- the leader sends version 1
Assigment
back as long as at least one version 1Subsription
is received - if the leader receives only version 2
Subscirption,
it send version 2Assigment
back
- bounced instance can send a version 2
- prepare a jar hot swap from 0.10.0.x to 1.2; Kafka Streams need to be configured with
- upgrade flow:
...