Status
Current state: Vote pending Accepted
Discussion thread: here
Vote thread: here
...
Change of the ELR does not require a leader epoch bump. In most cases, the ELR updates along with the ISR changes. The only case of the ELR changes alone is when an ELR broker registers after an unclean shutdown. In this case, no need to bump the leader epoch.
When updating the config min.insync.replicas, if the new min ISR <= current ISR, the ELR will be removed.
A new metric of Electable leaders will be added. It reflects the count of (ISR + ELR).
The AlterPartitionReassignments. The leader updates the ISR implicitly with AlterPartition requests. The controller will make sure than upon completion, the ELR only contains replicas in the final replica set. Additionally, in order to improve the durability of the reassignment
- The current behavior, when completing the reassignment, all the adding replicas should be in ISR. This behavior can result in 1 replica in ISR. Also, ELR may not help here because the removing ISR replicas can not stay in ELR when completed. So we propose to enforce that the reassignment can only be completed if the ISR size is larger or equal to min ISR.
- This min ISR requirement is also enforced when the reassignment is canceled.
Have a new admin API DescribeTopicRequest DescribeTopicsRequest for showing the topic details. We don't want to embed the ELR info in the Metadata API. The ELR is not some necessary details to be exposed to user clients.
- More public facing details will be discussed in the DescribeTopicRequest DescribeTopicsRequest section.
We also record the last-known ELR members.
It basically means when an ELR member has an unclean shutdown, it will be removed from ELR and added to the LastKnownELR. The LastKnownELR will be cleaned when ISR reaches the min ISR.
LastKnownELR is stored in the metadata log.
LastKnownELR will be also useful in the Unclean Recovery section.
- The last known leader will be tracked.
- This can be used if the Unclean recovery is not enabled. More details will be discussed in the Deliver Plan.
- The controller will record the last ISR member(the leader) when it is fenced.
- It will be cleaned when a new leader is elected.
...
During the shutdown, write the current broker epoch in the CleanShutdownFile.
During the start, the broker will try to read the broker epoch from the CleanShutdownFile. Then put this broker epoch in the broker registration request.
The controller will verify the broker epoch in the request with its registration record. If it is the same, it is a clean shutdown.
- if the broker shuts down before it receives the broker epoch, it will write -1.
Note, the CleanShutdownFile is removed after the log manager is initialized. It will be created and written when the log manager is shutting down.
Unclean recovery
As the new proposal allows the ISR to be empty, the leader election strategy has to be reviewed.
...
The kafka-leader-election.sh tool will be upgraded to allow manual leader election.
It can directly select a leader.
It can trigger an unclean recovery for the replica with the longest log in either Aggressive or Balance mode.
- Configs to update. Please refer to the Config Changes section
- For compatibility, the original unclean.leader.election.enable options True/False will be mapped to unclean.recovery.strategy options.
. Described in the above section. Balanced is the default value. - unclean.recoveryleader.managerelection.enabled. True for using the unclean recovery manager to perform an unclean recovery. False otherwise. False is the default value.
- unclean.recovery.timeout.ms. The time limits of waiting for the replicas' response during the Unclean Recovery. 5 min is the default value.
For compatibility, the original unclean.leader.election.enable options True/False will be mapped to unclean.recovery.strategy options.- unclean.leader.election.enable.false -> unclean.recovery.strategy.enable.false -> unclean.recovery.strategy.Balanced
- unclean.leader.election.enable.true -> unclean.recovery.strategy.Aggressive
...
{
"apiKey":62,
"type": "request",
"listeners": ["controller"],
"name": "BrokerRegistrationRequest",
"validVersions": "0-2",
"flexibleVersions": "0+",
"fields": [
...
// New fields begin.
{ "name": "PreviousBrokerEpoch", "type": "int64", "versions": "2+", "default": "-1",
"about": "The epoch before a clean shutdown." }
// New fields end.
]
}
|
...
DescribeTopicPartitionsRequest (Coming with ELR)
Should be issued by admin clients. The broker will serve this request. As a part of the change, the admin client will start to use DescribeTopicRequest
to query the topic, instead of using the metadata requests.
On the other hand, The following changes may affect the client side.
- The TopicPartitionInfo will also updated to include the ELR info.
- kafka-topics.sh does not have changes to its API but will have new fields in the output for the ELR, LastKnownELR, and LastKnownLeader.
ACL: Describe Topic
Limit: 20 topics max per request
More admin client related details please refer to the Admin API/Client changes
ACL: Describe Topic
The caller can list the topics interested or keep the field empty if requests all of the topics.
Pagination.
This is a new behavior introduced. The caller can specify the maximum number of partitions to be included in the response.
If there are more partitions than the limit, these partitions and their topics will not be sent back. In this case, the Cursor field will be populated. The caller can include this cursor in the next request.
Note,
- There is also a server-side config to control the maximum number of partitions to return. max.request.partition.size.limit
- There is no consistency guarantee between requests.
- It is an admin client facing API, so there is no topic id supported.
{
"apiKey": 74,
{
"apiKey":69,
"type": "request",
"listeners": ["broker"],
"name": "DescribeTopicRequestDescribeTopicPartitionsRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]stringTopicRequest", "versions": "0+",
"about": "The topics to fetch details for.", "versions": "0+", "entityType": "topicName"}
]
}
|
DescribeTopicResponse
"fields": [
{ "{
"apiKey":69,
"type": "request",
"name": "DescribeTopicResponseName",
"validVersionstype": "0string",
"flexibleVersionsversions": "0+",
"fields": [
{ "name
"about": "Topics", "type": "[]MetadataResponseTopic"The topic name", "versions": "0+",
"aboutentityType": "Each topic in the response.", "fields": [
topicName"}
]
},
{ "name": "ErrorCodeResponsePartitionLimit", "type": "int16int32", "versions": "0+", "default": ",
2000",
"about": "The topicmaximum error,number orof 0partitions ifincluded therein wasthe no errorresponse." },
{ "name": "NameCursor", "type": "stringCursor", "versions": "0+", "mapKeynullableVersions": true"0+", "entityTypedefault": "topicNamenull",
"nullableVersionsabout": "0+",
"about": "The topic name." },
The first topic and partition index to fetch details for.", "fields": [
{ "name": "TopicIdTopicName", "type": "uuidstring", "versions": "0+", "ignorable": true,
"about": "The topicname id." },
{ "name": "IsInternal", "type": "boolfor the first topic to process", "versions": "0+", "defaultentityType": "false", "ignorable": true,
"about": "True if the topic is internal." },
topicName"},
{ "name": "PartitionsPartitionIndex", "type": "[]MetadataResponsePartitionint32", "versions": "0+",
"about": "EachThe partition index into thestart with"}
]}
]
}
|
DescribeTopicsResponse
{
"apiKey": 74,
"typetopic.", "fields": [
{ "name": "ErrorCoderesponse",
"typename": "int16DescribeTopicPartitionsResponse",
"versionsvalidVersions": "0+",
"about
"flexibleVersions": "The partition error, or 0 if there was no error." },
0+",
"fields": [
{ "name": "PartitionIndexThrottleTimeMs", "type": "int32", "versions": "0+", ",
ignorable": true,
"about": "The partition indexduration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
{ "name": "LeaderIdTopics", "type": "int32[]DescribeTopicPartitionsResponseTopic", "versions": "0+", "entityType": "brokerId",
"
"about": "TheEach IDtopic ofin the leader broker." },
response.", "fields": [
{ "name": "LeaderEpochErrorCode", "type": "int32int16", "versions": "0+", "default": "-1", "ignorable": true,
"about": "The leader epoch of this partition topic error, or 0 if there was no error." },
{ "name": "ReplicaNodesName", "type": "[]int32string", "versions": "0+", "entityTypemapKey": true, "brokerIdentityType",
"about: "topicName", "nullableVersions": "The set of all nodes that host this partition"0+",
"about": "The topic name." },
{ "name": "IsrNodesTopicId", "type": "[]int32uuid", "versions": "0+", "entityTypeignorable": "brokerId",
true, "about": "The set of nodes that are in sync with the leader for this partitiontopic id." },
{ "name": "EligibleLeaderReplicasIsInternal", "type": "[]int32bool", "defaultversions": "null0+", "entityTypedefault": "brokerIdfalse",
"versionsignorable": "0+", "nullableVersions": "0+",
true,
"about": "TheTrue newif eligiblethe leadertopic replicasis otherwiseinternal." },
{ "name": "LastKnownELRPartitions", "type": "[]int32DescribeTopicPartitionsResponsePartition", "defaultversions": "null0+",
"entityType "about": "brokerIdEach partition in the topic.",
"fields": [
{ "name": "ErrorCode", "versionstype": "0+int16", "nullableVersionsversions": "0+",
"about": "The partition error, or 0 "about": "The last known ELRif there was no error." },
{ "name": "LastKnownLeaderPartitionIndex", "type": "int32", "defaultversions": "null0+",
"entityTypeabout": "brokerId",
The partition index." },
{ "name": "LeaderId", "type": "int32", "versions": "0+", "nullableVersionsentityType": "0+brokerId",
"about": "The ID lastof knownthe leader broker." },
{ "name": "OfflineReplicasLeaderEpoch", "type": "[]int32", "versions": "0+", "ignorabledefault": true"-1", "entityTypeignorable": "brokerId",
true,
"about": "The setleader of offlineepoch replicas of this partition." }
]},
{ "name": "TopicAuthorizedOperationsReplicaNodes", "type": "[]int32", "versions": "0+", "defaultentityType": "-2147483648brokerId",
"about": "32-bitThe bitfieldset toof representall authorizednodes operationsthat forhost this topicpartition." }
]
}
|
CleanShutdownFile (Coming with ELR)
,
{ "BrokerEpochname": "IsrNodes"xxx"}
|
ElectLeadersRequest (Coming with Unclean Recovery)
ACL: CLUSTER_ACTION
Limit: 20 topics per request.
{
"apiKey": 43,
"type, "type": "[]int32", "versions": "request0+",
"listenersentityType": ["zkBrokerbrokerId",
"brokerabout", "controller"],
: "The set of nodes that are in sync with the leader for this partition." },
{ "name": "ElectLeadersRequestEligibleLeaderReplicas",
"validVersionstype": "0-3[]int32",
"flexibleVersionsdefault": "2+null",
"fieldsentityType": [
...
{ "name": "TopicPartitions", "type": "[]TopicPartitions","brokerId",
"versions": "0+", "nullableVersions": "0+",
"about": "The topicnew partitionseligible toleader electreplicas leadersotherwise." },
"fields": [
...
// New fields begin.
{ "name": "DesiredLeadersLastKnownELR", "type": "[]int32", "versionsdefault": "3+null",
"about"entityType": "brokerId",
"versions": "0+"The desired leaders. The entry should matches with the entry in Partitions by the index, "nullableVersions": "0+",
"about": "The last known ELR." },
},
// New fields end.
] },
{ "name": "TimeoutMsOfflineReplicas", "type": "[]int32", "versions": "0+", "defaultignorable": "60000"true,
"aboutentityType": "brokerId"The time in ms to wait for the election to complete,
"about": "The set of offline replicas of this partition." }
]
}
|
GetReplicaLogInfo Request (Coming with Unclean Recovery)
ACL: CLUSTER_ACTION
Limit: 2000 partitions per request.
{
"apiKey":70,
"type,
{ "name": "requestTopicAuthorizedOperations",
"listenerstype": ["brokerint32"],
"nameversions": "GetReplicaLogInfoRequest0+",
"validVersionsdefault": "0-2147483648",
"flexibleVersionsabout": "0+",
"fields": [
32-bit bitfield to represent authorized operations for this topic." }]
},
{ "name": "BrokerIdNextTopicPartition", "type": "int32Cursor", "versions": "0+", "entityTypenullableVersions": "brokerId0+",
"default": "null",
"about": "The ID of the broker." },
next topic and partition index to fetch details for.", "fields": [
{ "name": "TopicPartitionsTopicName", "type": "[]TopicPartitionsstring", "versions": "0+", "nullableVersions": "0+",
"about": "The topicname partitionsfor tothe elect leaders.",
"fields": [
{ "name": "TopicId", "type": "uuid"first topic to process", "versions": "0+", "ignorableentityType": true, "about": "The unique topic ID"topicName"},
{ "name": "PartitionsPartitionIndex", "type": "[]int32", "versions": "0+",
"about": "The partitionspartition ofindex thisto topic whose leader should be elected." },
]}
] }
|
GetReplicaLogInfo Response
CleanShutdownFile (Coming with ELR)
It will be a JSON file.
{
"version": 0
"BrokerEpoch":"xxx"
}
|
ElectLeadersRequest (Coming with Unclean Recovery)
ACL: CLUSTER_ACTION
Limit: 1000 partitions per request. If more than 1000 partitions are included, only the first 1000 will be served. Others will be returned with REQUEST_LIMIT_REACHED.
{
"apiKey": 43,
"type": "request{
"apiKey":70,
"type": "response",
"name": "GetReplicaLogInfoResponse",
"validVersions": "0",
"flexibleVersionslisteners": ["0+zkBroker",
"fieldsbroker": [, "controller"],
{ "name": "BrokerEpochElectLeadersRequest",
"typevalidVersions": "int640-3",
"versionsflexibleVersions": "02+",
"aboutfields": "The[
epoch for the broker." } ...
{ "name": "LogInfoListTopicPartitions", "type": "[]LogInfoTopicPartitions", "versions": "0+", "nullableVersions": "0+",
"about": "The listtopic ofpartitions theto logelect infoleaders.",
"fields": [
...
// New fields begin. The same level with the Partitions
{ "name": "TopicIdDesiredLeaders", "type": "uuid[]int32", "versions": "03+", "ignorablenullableVersions": true, "3+",
"about": "The uniquedesired topic ID."},
{ "name": "Partition", "type": "int32", "versions": "0+", "about": "The id for the partition." },
leaders. The entry should match with the entry in Partitions by the index." },
},
// New fields end.
] },
{ "name": "LastWrittenLeaderEpochTimeoutMs", "type": "int32", "versions": "0+", "aboutdefault": "60000"The,
last written leader epoch in the log"about": "The time in ms to wait for the election to complete." },
]
}
|
GetReplicaLogInfo Request (Coming with Unclean Recovery)
ACL: CLUSTER_ACTION
Limit: 2000 partitions per request. If more than 1000 partitions are included, only the first 1000 will be served. Others will be returned with REQUEST_LIMIT_REACHED.
{
"apiKey":70,
{ "nametype": "CurrentLeaderEpochrequest",
"typelisteners": ["int32broker"],
"versionsname": "0+GetReplicaLogInfoRequest",
"aboutvalidVersions": "0"The,
current leader epoch for the partition from the broker point of view." },"flexibleVersions": "0+",
"fields": [
{ "name": "LogEndOffsetBrokerId", "type": "int64int32", "versions": "0+", "aboutentityType": "The log end offset for the partition." }
]}
] }
|
kafka-leader-election.sh (Coming with Unclean Recovery)
...
// Updated field starts.
--election-type <[PREFERRED, UNCLEAN, LONGEST_LOG_AGGRESSIVE, LONGEST_LOG_BALANCED, DESIGNATION]:
"brokerId",
"about": "The ID of the broker." },
{ "name": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+", "nullableVersions": "0+",
"about": "The topic partitions to query the log info.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID"},
{ "name": "Partitions", "type": "[]int32", Type of election to attempt. Possible"versions": "0+",
election type> "about": "The partitions of this topic whose leader should be elected." },
]}
] }
|
GetReplicaLogInfo Response
{
"apiKey":70,
values are"type": "response",
"preferredname" for preferred: "GetReplicaLogInfoResponse",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "BrokerEpoch", "type": "int64", "versions": "0+", "about": "The epoch for the broker." }
{ "name": "TopicPartitionLogInfoList", leader election, or "unclean" for"type": "[]TopicPartitionLogInfo", "versions": "0+",
"about": "The list of the log info.",
"fields": [
{ "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID."},
{ a random unclean leader election,
"name": "PartitionLogInfo", "type": "[]PartitionLogInfo", "versions": "0+", "about": "The log info of a partition.",
"fields": [
{ "name": "Partition", "type": "int32", "versions": "0+", "about": "The id for the partition." },
{ or "longest_log_agressive"/"longest_log_balanced"
"name": "LastWrittenLeaderEpoch", "type": "int32", "versions": "0+", "about": "The last written leader epoch in the log." },
{ "name": "CurrentLeaderEpoch", "type": "int32", "versions": "0+", "about": "The current leader epoch for the partition from the broker point of view." },
to choose the replica
{ "name": "LogEndOffset", "type": "int64", "versions": "0+", "about": "The log end offset for the partition." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+", "about": "The result error, or zero if there was no error."},
{ with the longest log or "designation" for
"name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "about": "The result message, or null if there was no error."} ]},
]}
] }
|
kafka-leader-election.sh (Coming with Unclean Recovery)
...
// Updated field starts.
--election-type <[PREFERRED, UNCLEAN, LONGEST_LOG_AGGRESSIVE, LONGEST_LOG_BALANCED, DESIGNATION]: electing the given replica to be
the leader. If
Type preferredof election isto selection,attempt. thePossible
election type> values are
election is only performed if the
"preferred" for preferred leader election
current leader is not the preferred
or "unclean" for a random unclean leader election,
leader for the topic partition. If
or "longest_log_agressive"/"longest_log_balanced" to choose the replica
longest_log_agressive/longest_log_balanced/designation
with the longest log
election is selected, the
or "designation" for electing the given replica("desiredLeader") to electionbe isthe onlyleader. performed
if there
If preferred areelection nois leaderselection, for the
topic
election partition.is REQUIRED.only performed if the
--path-to-json-file <String: Path to Thecurrent JSONleader fileis withnot the list ofpreferred
JSON file> partition for which leader elections
leader for the topic partition. If
should be performed. This is an
longest_log_agressive/longest_log_balanced/designation
example format. The desiredLeader field
election is selected, the
is only required in DESIGNATION election.
is only performed if there
are no leader for the topic
{"partitions":
partition. REQUIRED. [{"topic": "foo", "partition": 1, "desiredLeader": 0},
--path-to-json-file <String: Path to The JSON file with the list of
JSON file> {"topic": "foobar", "partition": 2, "desiredLeader": 1}]
partition for which leader elections
should }
be performed. This is an
example Notformat. allowedThe if --all-topic-partitionsdesiredLeader field
or --topic flags are specified.
// Updated field endsis only required in DESIGNATION election.
{"partitions":
[{"topic": "foo", "partition": 1, "desiredLeader": 0},
{"topic": "foobar", "partition": 2, "desiredLeader": 1}]
}
Not allowed if --all-topic-partitions
or --topic flags are specified.
// Updated field ends.
|
|
Config changes
The new configs are introduced for ELR
- eligible.leader.replicas.enabled. It controls whether the controller will record the ELR-related metadata and whether ISR can be empty. False is the default value. It will turn true in the future.
- max.request.partition.size.limit. The maximum number of partitions to return in a API response.
The new configs are introduced for Unclean Recovery.
- unclean.recovery.strategy. Described in the above section. Balanced is the default value.
- unclean.recovery.manager.enabled. True for using the unclean recovery manager to perform an unclean recovery. False means the random election will be used in the unclean leader election. False is the default value.
- unclean.recovery.timeout.ms. The time limits of waiting for the replicas' response during the Unclean Recovery. 5 min is the default value.
New Errors
REQUEST_LIMIT_REACHED
As we introduce the request limit for the new APIs, the items sent in the request but over the limit will be returned with REQUEST_LIMIT_REACHED. It is a retriable error.
Admin API/Client changes
The admin client will start to use the DescribeTopicsRequest to describe the topic.
- The client will split a large request into proper pieces and send them one after another if the requested topics count reaches the limit.
- The client will retry querying the topics if they received the response with Cursor field.
- The output of the topic describe will be updated with the ELR related fields.
- TopicPartitionInfo will be updated to include the ELR related fields.
Metrics
The following metrics will be added for ELR
...
min.insync.replicas will no longer be effective to be larger than the replication factor. For existing configs, the min.insync.replicas will be min(min.insync.replicas, replication factor).
Cluster admin should update the min.insync.replicas to 1if they want to have the replication going when there is only the leader in the ISR.
- Note that, this new requirement is not guarded by any feature flags/Metadata version.
ELR
It will be guarded by a new metadata version and the eligible.leader.replicas.enabled. So it is not enabled during the rolling upgrade.
After the controller picked up the new MV and eligible.leader.replicas.enabled is true, when it loads the partition states, it will populate the ELR as empty if the PartitionChangeRecord uses an old version. In the next partition update, the controller will record the current ELR.
MV downgrade: Once the MV version is downgraded, all the ELR related fields will be removed on the next partition change. The controller will also ignore the ELR fields.
Software downgrade: After the MV version is downgraded, a metadata delta will be generated to capture the current status. Then the user can start the software downgradeNote, the leader can only enable empty ISR after the new metadata version.
Clean shutdown
It will be guarded by a new metadata version. Before that, the controller will treat all the registrations with unclean shutdowns.
...