Table of Contents |
---|
Status
Current state: "Under Discussion"Accepted (vote)
Discussion thread: TODO [DISCUSS] KIP-268: Simplify Kafka Streams Rebalance Metadata Upgrade
JIRA: KAFKA-6054
Released: 12.20
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
Kafka Streams exchanges custom metadata during rebalance. This metadata has a magic version number encode to allow upgrade upgrading of the metadata format. In the past we upgrade upgraded the metadata format once (in 0.10.1.0 release), but did not design a proper upgrade path. This implies, that upgrading from 0.10.0.x versions, is only possible if the whole application is shut down first (ie, rolling bounces don't work).
In 12.20, we plan to upgrade the metadata format again and thus need to fix KAFKA-6054. The current implementation forces us, to design an upgrade path with two rolling bounced. This also requires that users configure the application correctly during upgrade (see details below). This is cumbersome and error prone for the user. Thus, we also propose a simplified upgrade path with no requirements for users to configure Kafka Stream Streams correctly for upgrade and also allows to allow for single rolling bounce upgradeupgrades.
Note, that this KIP is requirement for KIP-258 and KIP-262.
...
For the upgrade from 0.10.0.x, ..., 1.1.x version to 12.2 0 version, we need to add a new configuration parameter upgrade.from
that will be null
by default and can take the following values:
null
: no upgrade required (if a new application is started or upgrade was done already)"0.10.0.x"
,"0.10.1.x"
,"0.10.2.x"
,"0.11.0.x"
,"1.0.x"
, and"1.1.x"
: for upgrading from 0.10.0.x,..., 1.1.x to 12.20
Note, that the above proposal only fixes KAFKA-6054 in 12.20. If we want to have fixes for versions 0.10.1.x, ...,1.1.x for KAFKA-6054, we would need to back port only one required valuesvalue:
null
: for no upgrade required"0.10.0.x"
: for upgrading from 0.10.0.x to any version of 0.10.1.x, ..., 1.1.x
...
We add the above config for upgrading from 0.10.0.x, ..., 1.1.x to 12.2 0 and future releases. We describe the upgrade path in detail below. To enable single rolling bounce upgrade with no required configuration of the applications, we also add a second magic byte encoding "supported version" into the rebalance metadata that allows us to implement a "version probing" step. If we want to change the metadata in pre-12.2 0 releases again, this "version probing" step allows for single rolling bounce upgrade (details below).
Note, that the currently used rebalance metadata version are 1 (0.10.0.x) and 2 (0.10.1.x, ..., 1.1.x). We increase the metadata version numer number to 3 in 12.2 release.
...
0 release.
Code Block | ||
---|---|---|
| ||
// Metadata v2
SubscriptionMetadata => VersionNumber ClientId PreviousAssignedTasks StandbyTasks UserEndpoint
VersionNumber => int32
ClientId => int64 int64 // UUID most signification bits + least significant bits
PreviousAssignedTasks => [TaskId]
StandbyTasks => [TaskId]
UserEndpoint => string // user specified endpoint for Interactive Queries
AssignmentMetadata => VersionNumber AssignedTasks AssignedStandbyTasks GlobalAssignment
VersionNumber => int32
AssignedTasks => [TaskId]
AssignedStandbyTasks => [TaskId]
GlobalAssignment => [Host Port [Topic Partition]]] // metadata for Interactive Queries
TaskId => GroupId Partition
GroupId => int32
Partition => int32
Host => string
Port => int32
Topic => string
Partitio => int32
// Metadata v3
SubscriptionMetadata => VersionNumber SupportedVersionNumber ClientId PreviousAssignedTasks StandbyTasks UserEndpoint
VersionNumber => int32
SupportedVersionNumber => int32 // NEW
ClientId => int64 int64
PreviousAssignedTasks => numTasks [TaskId]
StandbyTasks => numTasks [TaskId]
UserEndpoint => string
AssignmentMetadata => VersionNumber SupportedVersionNumber AssignedTasks AssignedStandbyTasks GlobalAssignment
VersionNumber => int32
SupportedVersionNumber => int32 // NEW
AssignedTasks => [TaskId]
AssignedStandbyTasks => [TaskId]
GlobalAssignment => [Host Port [Topic Partition]]] |
Upgrading to 2.0:
- prepare a jar hot swap from old version to 12.20; Kafka Streams need to be configured with
upgrade.from="<old.version>"
for startup - do a rolling bounce to get the new jar and config in place for each instance
- config
"upgrade.from"
tells the application to sendSubscription
using old endocing (version 1 or 2), to be compatible with potential version 1 or 2 leader in the group - instances will receive a corresponding version 1 or 2
Assigment
in this stage
- config
- user prepares a second round of rebalance; this time, the configuration parameter
upgrade.from
must be removed for new startup - do a second rolling bounce for each instance to get new config (ie,
upgrade.from
removed)- bounced instance can send a version 3
Subscription
as we know that leade will undestand leader understands version 3 - the leader sends version 1 or 2
Assigment
back as long as at least one version 1 or 2Subsription
is received - if the leader receives only version 3
Subscirption,
it send version 3Assigment
and the upgrad is completed
- bounced instance can send a version 3
Future upgrades with "version probing" (ie, upgrading from 2.0 to later release — only 'triggers' it metadata version number is increased):
In the current implemenationimplementation, the group leader fails if it receives a subscription with a higher version number than it understands. We propose to change this: instead of failing, the leader will send an empty assignment back encoding its supported version. This allows the upgrade upgraded follower to downgrade it its subscription and rejoin the group sending a subscription that the (not yet) upgraded leader understands. As we always encode the leader's supported version in the assignment, after the leader is upgrade and understand the new metadata version, all other instances can switch back to the higherst supporte highest supported metadata version.
Detailed upgrade protocol from metadata version X to Y (with X >= 2.0):
- On startup/rolling-bounce, an instance does not know what version the leader understands and (optimistically) sends an
Subscription
with the latest version Y - (Old, ie, not yet upgraded) Leader sends empty
Assignment
back to the corresponding instance that sent the newerSubscription
it does not understand. TheAssignmet
Assignment
metadata only encods encodes both version numbers (used-version == supported-version) as leader's supported-version X. - For all other instances the leader sends a regular
Assignment
in version X back. - If an upgrade follower sends new version number Y
Subscription
and receives version XAssignment
with "supported-version = X", it can downgrade to X (in-memory flag) and resends a newSubscription
with old version X to retry joining the group. To force a an immediate second rebalance, the follower does an "unsubscribe()/subscribe()/poll()" sequence. - As long as the leader (before or after upgrade) receives at least one old version X
Subscription
it always send used-sends versionAssignment
X back (the encoded supported version is X before the leader is upgrade and Y after the leader is upgraded). - If an upgraded instance receives an
Assigment
it always checks the leaders supported-version and update its downgraded "used-version" if possible
Thus, a single rolling bounce without any configs setting config settings is required sufficient as the leader allows to probe its supported version instead of failing. Note, that if the leader is bounced last, the metadata upgrad upgrade only finishes after one more rebalance. We can trigger this rebalance with one more "unsubscribe()/subscribe()/poll()" sequnce sequence (to make sure only one instance executes this, the leader should be responsible to trigger this final rebalance – note, if the leader is not bounced last, we can detect this and avoid the additional rebalance).
...
Increasing the rebalance metadata version to 3 is not a backward compatible change per-se. However, the outlined upgrade path allows userd users to upgrade to 12.2 0 with zero downtime and two rolling bounces. Note, that an simplified "offline upgrade" is also possible. Instead of setting configs and doing two rolling bouncedbounces, all application instances are can be stopped. Afterwards, all instances are restarted with the new 12.2 0 jar.
Test Plan
- unit and integration tests for StreamPartitionsAssigner to that must react correctly to configs and received subscription versions
- system tests that perform rolling bounce upgrades as described above
- this should include failure scenario during the upgrade
- this should include "simulated upgrades" to metadata version 4, to ensure that the implementation implemented "version probing" work correct for future changes
Test matrix:
from version | to 0.10.1.x | to 0.10.2.x | to 0.11.0.x | to 1.0.x | to 1.1.x | to 2.0.x | to post-2.0.x (simulate metadata version 4) |
---|---|---|---|---|---|---|---|
0.10.0.x | x (*) | x (*) | x (*) | x (*) | x (*) | x | x |
0.10.1.x | x (*) | x (*) | x (*) | x (*) | x | x | |
0.10.2.x | x (*) | x (*) | x (*) | x | x | ||
0.11.0.x | x (*) | x (*) | x | x | |||
1.0.x | x (*) | x | x | ||||
1.1.x | x | x | |||||
2.0.x | x (tests "version probing") |
(*): requires back porting of KAFKA-6054 to older branches
Rejected Alternatives
- use consumer's built-in protocal upgrade mechanism (ie, register multple "assigment strategies")
- has the disadvantage that we need to implement two
StreamsPartitionAssingor
classes - increased network traffic during rebalance
- encoding "supported version" in metadata (ie, version probing) subsumes this approach for future releases
- if we want to "disable" the old protocol, a second rebalance is required, too
- has the disadvantage that we need to implement two
- Don't fix KAFKA-6054
- it's a simple fix to include: just add one more accepted values to parameter
upgrade.from
- it's s fair question, how many people will run with Streams 0.10.0 – note thosethought, that if people are "stuck" with 0.10.0 broker, they can cannot use 0.10.1 or newer as it's not backwards compatible to 0.10.0 – thus, might be more than expected
- it's a simple fix to include: just add one more accepted values to parameter
- Fix KAFKA-6054 only for 12.2 0 release
- it's a relativley relatively simply fix for older releases (main desing design work is already coverd covered and writing the code is not to complex becuase it's only the rebalance metadata version changetoo complex)
- it's unclear though if we will have bug-fix releases for older versions; thus nobody might ever be able to get this code (if they don't build from corresponding dev-branches themselves)
...