Table of Contents

Status

Current state: Accepted

Under Discussion thread: here

Discussion Vote thread: here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

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
1. 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.
2. 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.
1. More public facing details will be discussed in the DescribeTopicRequest DescribeTopicsRequest section.
We also record the last-known ELR members.
1. 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.
2. LastKnownELR is stored in the metadata log.
3. LastKnownELR will be also useful in the Unclean Recovery section.
The last known leader will be tracked.
1. This can be used if the Unclean recovery is not enabled. More details will be discussed in the Deliver Plan.
2. The controller will record the last ISR member(the leader) when it is fenced.
3. 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.

...

Randomly electing a leader is definitely worth improving. As a result, we decide to replace the unclean leader random election with the Unclean Recovery.

The Unclean Recovery uses a deterministic way to elect the leader persisted the most data. On a high level, once the unclean recovery is triggered, the controller will use a new API GetReplicaLogInfo to query the log end offset and the leader epoch from each replica. The one with the highest leader epoch plus the longest log end offset will be the new leader. To help explain when and how the Unclean Recovery is performed, let's first introduce some config changes.

The current unclean.leader.election.enable will be replaced by an intent-based config, new unclean.recovery.strategy has the following 3 options.

ProactiveAggressive. It represents the intent of recovering the availability as fast as possible.
Balanced. Auto recovery on potential data loss case, wait as needed for a better result.
ManualNone. Stop the partition on potential data loss.

...

If there are other ISR members, choose an ISR member.
If there are unfenced ELR members, choose an ELR member.
If there are fenced ELR members
1. If the unclean.recovery.strategy=ProactiveAggressive, then an unclean recovery will happen.
2. Otherwise, we will wait for the fenced ELR members to be unfenced.
If there are no ELR members.
1. If the unclean.recovery.strategy=ProactiveAggressive, the controller will do the unclean recovery.
2. If the unclean.recovery.strategy=Balanced, the controller will do the unclean recovery when all the LastKnownELR are unfenced. See the following section for the explanations.
3. Otherwise, unclean.recovery.strategy=ManualNone, the controller will not attempt to elect a leader. Waiting for the user operations.

...

1. .

The controller will initiate the Unclean Recovery once a partition meets the above conditions. As mentioned above, the controller collects the log info from the replicas. Apart from the Log end offset and the partition leader epoch in the log, the replica also returns the following for verification:

...

In Balance mode, all the LastKnownELR members have replied, plus the replicas replied within the timeout. Due to this requirement, the controller will only start the recovery if the LastKnownELR members are all unfenced.
In Aggressive Proactive mode, any replicas replied within a fixed amount of time OR the first response received after the timeout.

...

The kafka-leader-election.sh tool will be upgraded to allow manual leader election.
1. It can directly select a leader.
2. It can trigger an unclean recovery for the replica with the longest log in either Proactive 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. add
1. unclean.recovery.strategy. Described in the above section. Balanced is the default value.
2. unclean.recovery.Enabled. True for enabling the unclean recovery. False otherwise. False is the default value.
3. unclean.recovery.timeout.ms. The time limits of waiting for the replicas' response during the Unclean Recovery. 5 min is the default value.
For a better user experience, the unclean.recovery.strategy and
1. unclean.leader.election.enable.false -> unclean.recovery.strategy.Balanced
1. unclean.leader.election.enable
will be converted if unclean
1. .
recovery.Enabled is changed.
1. unclean.recovery.Enabled from false to true
  unclean.leader.election.enabletrue -> unclean.recovery.strategy falseBalancedtrueProactive
  unclean.recovery.Enabled from true to false
  unclean.recovery.strategy unclean.leader.election.enableProactivetrueBalancedfalseManualfalsestrategy.Aggressive

Public Interfaces

We will deliver the KIP in phases, so the API changes are also marked coming with either ELR or Unclean Recovery.

...

{
  "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.",
      "versionsfields": "0+", "entityType[
        { "name": "topicName"}
]
}

DescribeTopicResponse

{
  "apiKey":69,
  Name", "type": "requeststring",
   "nameversions": "DescribeTopicResponse0+",
  "validVersions
          "about": "0The topic name",
   "flexibleVersionsversions": "0+",
   "fieldsentityType": [
    { "name": "TopicstopicName"}
      ]
    },
    { "name": "ResponsePartitionLimit", "type": "[]MetadataResponseTopicint32", "versions": "0+",
      ", "default": "2000",
      "about": "Each topicThe maximum number of partitions included in the response." }, "fields": [
      
    { "name": "ErrorCodeCursor", "type": "int16Cursor", "versions": "0+",
         "aboutnullableVersions": "0+"The topic error, or 0 if there was no error." },
      , "default": "null",
      "about": "The first topic and partition index to fetch details for.", "fields": [
      { "name": "NameTopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType
        "about": "topicNameThe name for the first topic to process", "nullableVersionsversions": "0+",
         "aboutentityType": "The topic name." },
      "topicName"},
      { "name": "TopicIdPartitionIndex", "type": "uuidint32", "versions": "0+", "ignorable": true, "about": "The topicpartition index to start with"}
    ]}
  ]
}

DescribeTopicsResponse

{
  "apiKey": 74,
 id." },
      { "name": "IsInternal", "type": "boolresponse",
  "versionsname": "0+DescribeTopicPartitionsResponse",
  "defaultvalidVersions": "false0",
  "ignorableflexibleVersions": true,
        "about": "True if the topic is internal." },
      "0+",
  "fields": [
    { "name": "PartitionsThrottleTimeMs", "type": "[]MetadataResponsePartitionint32", "versions": "0+",
         "aboutignorable": "Each partition in the topic.", "fields": [
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The partition error, or 0 if there was no error." },
        true,
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "PartitionIndexTopics", "type": "int32[]DescribeTopicPartitionsResponseTopic", "versions": "0+",
          
      "about": "TheEach partitiontopic index." },
        in the response.", "fields": [
      { "name": "LeaderIdErrorCode", "type": "int32int16", "versions": "0+", "entityType": "brokerId",
          
        "about": "The ID of the leader broker topic error, or 0 if there was no error." },
        
      { "name": "LeaderEpochName", "type": "int32string", "versions": "0+", "mapKey": true, "defaultentityType": "-1topicName", "ignorablenullableVersions": true,
          "0+",
        "about": "The leadertopic epoch of this partition." },
        name." },
      { "name": "ReplicaNodesTopicId", "type": "[]int32uuid", "versions": "0+", "entityTypeignorable": "brokerId",
          true, "about": "The set of all nodes that host this partitiontopic id." },
        
      { "name": "IsrNodesIsInternal", "type": "[]int32bool", "versions": "0+", "entityTypedefault": "brokerIdfalse",
           "aboutignorable": "The set of nodes that are in sync with the leader for this partitiontrue,
        "about": "True if the topic is internal." },
           { "name": "EligibleLeaderReplicasPartitions", "type": "[]int32DescribeTopicPartitionsResponsePartition", "defaultversions": "null0+",
        "entityTypeabout": "brokerId",
          "versionsEach partition in the topic.", "fields": [
        { "name": "ErrorCode", "type": "0+int16", "nullableVersionsversions": "0+",
         
          "about": "The new eligible leader replicas otherwise partition error, or 0 if there was no error." },

            { "name": "LastKnownELRPartitionIndex", "type": "[]int32", "defaultversions": "null0+",
          "entityTypeabout": "brokerId",
         The partition index." },
        { "name": "LeaderId", "type": "int32", "versions": "0+", "nullableVersionsentityType": "0+brokerId",

               "about": "The ID of lastthe knownleader ELRbroker." },
            { "name": "LastKnownLeaderLeaderEpoch", "type": "int32", "defaultversions": "null0+", "entityTypedefault": "brokerId-1",
 "ignorable": true,
             "versionsabout": "0+", "nullableVersions": "0+",
          "about": "The last known leaderThe leader epoch of this partition." },
         { "name": "OfflineReplicasReplicaNodes", "type": "[]int32", "versions": "0+", "ignorable": true, "entityType": "brokerId",
          
          "about": "The set of all offlinenodes replicasthat ofhost this partition." }
      ]},
        { "name": "TopicAuthorizedOperationsIsrNodes", "type": "[]int32", "versions": "0+", "defaultentityType": "-2147483648brokerId",
             "about": "32-bitThe bitfieldset toof representnodes authorized operationsthat are in sync with the leader for this topicpartition." }
]
}

CleanShutdownFile (Coming with ELR)

,
        { "BrokerEpoch":"xxx"}

ElectLeadersRequest (Coming with Unclean Recovery)

ACL: CLUSTER_ACTION

Limit: 20 topics per request.

{
  "apiKey": 43,
 name": "EligibleLeaderReplicas", "type": "request[]int32",
  "listenersdefault": ["zkBrokernull", "brokerentityType",: "controllerbrokerId"],
 ,
          "nameversions": "ElectLeadersRequest0+",
  "validVersionsnullableVersions": "0-3+",
  "flexibleVersions
          "about": "2+",
  "fields": [
  ...
 The new eligible leader replicas otherwise." },
        { "name": "TopicPartitionsLastKnownELR", "type": "[]TopicPartitions", int32", "default": "null", "entityType": "brokerId",
          "versions": "0+", "nullableVersions": "0+",

          "about": "The topiclast partitions to elect leadersknown ELR.",
    "fields": [
    ...

// New fields begin.
  { "name": "DesiredLeaders", "type": "[]int32", "versions": "3+",
       },
        { "name": "OfflineReplicas", "type": "[]int32", "versions": "0+", "ignorable": true, "entityType": "brokerId",
          "about": "The desiredset leaders.of Theoffline entryreplicas shouldof matches with the entry in Partitions by the indexthis partition." },
  ]},
// New fields end.

  ] },
  
      { "name": "TimeoutMsTopicAuthorizedOperations", "type": "int32", "versions": "0+", "default": "60000-2147483648",

        "about": "The32-bit timebitfield into msrepresent toauthorized waitoperations for the election to completethis topic." }
  ] 
}

GetReplicaLogInfo Request (Coming with Unclean Recovery)

ACL: CLUSTER_ACTION

Limit: 2000 partitions per request.

{
  "apiKey":70,
 ]
    },
    { "name": "NextTopicPartition", "type": "requestCursor",
  "listenersversions": ["broker0+"],
  "namenullableVersions": "GetReplicaLogInfoRequest0+",
  "validVersionsdefault": "0null",

      "flexibleVersionsabout": "0+The next topic and partition index to fetch details for.",
  "fields": [

       { "name": "BrokerIdTopicName", "type": "int32string", "versions": "0+",
        "entityTypeabout": "brokerId",The 
name for the first topic to process",  "aboutversions": "0+"The ID of the broker." },
    , "entityType": "topicName"},
      { "name": "TopicPartitionsPartitionIndex", "type": "[]TopicPartitionsint32", "versions": "0+", "nullableVersions": "0+",
    "about": "The topicpartition partitionsindex to electstart with"}
    ]}
  ]
}

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,
  leaders.",
    "fields": [
      { "name": "TopicId", "type": "uuidrequest",
  "versionslisteners": "0+", "ignorable": true, "about": "The unique topic ID"},
      ["zkBroker", "broker", "controller"],
  "name": "ElectLeadersRequest",
  "validVersions": "0-3",
  "flexibleVersions": "2+",
  "fields": [
  ...
  { "name": "PartitionsTopicPartitions", "type": "[]int32TopicPartitions", "versions": "0+",
    "nullableVersions": "0+",
    "about": "The partitions of this topic whosepartitions leaderto shouldelect be electedleaders." },
    ]}
] }

GetReplicaLogInfo Response

{
  "apiKey":70,
  "type": "response",
  "name": "GetReplicaLogInfoResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
 "fields": [
    ...

// New fields begin. The same level with the Partitions
     { "name": "BrokerEpochDesiredLeaders", "type": "int64[]int32", "versions": "03+", "aboutnullableVersions": "3+"The,
 epoch  for  the  broker."about": }
"The desired leaders. The { "name": "LogInfoList", "type": "[]LogInfo", "versions": "0+", 
    "about": "The list of the log info.",
    "fields": [
    entry should match with the entry in Partitions by the index." },
     },
// New fields end.

  ] },
  { "name": "TopicIdTimeoutMs", "type": "uuidint32", "versions": "0+", "ignorabledefault": true,"60000",
    "about": "The uniquetime 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,
  "type": "request",
  "listeners": ["broker"],
 topic ID."},
      { "name": "Partition", "type": "int32", "versions": "0+", "about": "The id for the partition." },
      { "name": "LastWrittenLeaderEpochGetReplicaLogInfoRequest",
  "typevalidVersions": "int320",
  "versionsflexibleVersions": "0+",
  "aboutfields": "The last written leader epoch in the log." },
  [
    { "name": "CurrentLeaderEpochBrokerId", "type": "int32", "versions": "0+", "aboutentityType": "ThebrokerId", 
 current leader epoch for the partition from the broker point"about": "The ID of the viewbroker." },
      { "name": "LogEndOffsetTopicPartitions", "type": "int64[]TopicPartitions", "versions": "0+", "nullableVersions": "0+",
    "about": "The logtopic endpartitions offsetto forquery the log partition." }
info.",
    "fields": [
      ]}
] }

kafka-leader-election.sh (Coming with Unclean Recovery)

...
// Updated field starts.
--election-type <[PREFERRED, UNCLEAN, LONGEST_LOG_PROACTIVE, LONGEST_LOG_BALANCED, DESIGNATION]:    { "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID"},
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
                    "about": "The partitions of this topic whose leader should be elected." },
    ]}
] }

GetReplicaLogInfo Response

{
  "apiKey":70,
  "type": "response",
  "name": "GetReplicaLogInfoResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
   Type of election to attempt. Possible"fields": [
  election type> { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "about": "The epoch for the broker." }
    { "name": "TopicPartitionLogInfoList", "type":     values are "preferred" for preferred"[]TopicPartitionLogInfo", "versions": "0+", 
    "about": "The list of the log info.",
    "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The unique topic ID."},
      {  leader election, or "unclean" for
         "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." },
      {  a random unclean leader election,
          "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." },
      or{ "longest_log_proactive"/"longest_log_balanced"
                                  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."},
      {  to choose the replica 
             "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]:          with the longest log or "designation" for
                                          electingType theof given replicaelection to beattempt. thePossible
 leader. If
election type>                          values are 
             preferred election is selection, the
                         "preferred" for preferred leader election
             election is only performed if the
                        or "unclean" for a random unclean leader election,
           current leader is not the preferred
                          or "longest_log_agressive"/"longest_log_balanced" to choose the replica 
          leader for the topic partition. If
                            with the longest log 
          longest_log_proactive/longest_log_balanced/designation 
                               or "designation" for electing the given replica("desiredLeader") to be the leader. election
 is selected, the
                                       If preferred  election is onlyselection, performedthe
 if there
                                        election is areonly noperformed leader forif the topic
                                          partition.current REQUIRED.leader is not the preferred
                                  
--path-to-json-file <String: Path to    The JSONleader file withfor the listtopic partition. ofIf
    JSON  file>                              partition for which leader elections
  longest_log_agressive/longest_log_balanced/designation 
                                       should be performed. Thiselection is selected, anthe
                                          exampleelection format. The desiredLeader fieldis only performed if there
                                          isare onlyno requiredleader infor DESIGNATIONthe election.topic
                                         
 partition. REQUIRED.                                      {"partitions":
   
--path-to-json-file <String: Path to    The JSON file with the list  of
  JSON file>                       	[{"topic": "foo", "partition": 1, "desiredLeader": 0},
  partition for which leader elections
                                  	 {"topic": "foobar", "partition": 2, "desiredLeader": 1}]
  should be performed. This is an
                                    }      example format. The desiredLeader field
                                        Not allowed if --all-topic-partitions
  is only required in DESIGNATION election.
                                        
                                        {"partitions":
                 or --topic flags are specified.
// Updated field ends                     	[{"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

...

The High Watermark will only advance if all the messages below it have been replicated to at least min.insync.replicas ISR members.
The consumer is affected when consuming the ack=1 and ack=0 messages. When there is only 1 replica(min ISR=2), the HWM advance is blocked, so the incoming ack=0/1 messages are not visible to the consumers. Users can avoid the side effect by updating the min.insync.replicas to 1 for their ack=0/1 topics.
Compared to the current model, the proposed design has availability trade-offs:
1. If the network partitioning only affects the heartbeats between a follower and the controller, the controller will kick it out of ISR. If losing this replica makes the ISR under min ISR, the HWM advancement will be blocked unnecessarily because we require the ISR to have at least min ISR members. However, it is not a regression compared to the current system at this point. But later when the network partitioning finishes, the current leader will put the follower into the pending ISR(aka "maximum ISR") and continue moving forward while in the proposed world, the leader needs to wait for the controller to ack the ISR change.
2. Electing a leader from ELR may mean choosing a degraded broker. Degraded means the broker can have a poor performance in replication due to common reasons like networking or disk IO, but it is alive. It can also be the reason why it fails out of ISR in the first place. This is a trade-off between availability and durability.
The unclean leader election will be replaced by the unclean recovery.
For fsync users, the ELR can be beneficial to have more choices when the last known leader is fenced. It is worth mentioning what to expect when ISR and ELR are both empty. We assume fsync users adopt the unclean.leader.election.enable as false.
1. If the KIP has been fully implemented. The unclean.recovery.strategy will be balanced. During the unclean recovery, the controller will elect a leader when all the LastKnownElr members have replied.
2. If only the ELR is implemented, the LastKnownLeader is preferred when ELR and ISR are both empty.

...

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. 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.

...

Unclean Recovery is guarded by the feature flag unclean.recovery.Enabled. manager.enabled.

For the existing unclean.leader.election.enable

...

1. If true, unclean.recovery.strategy will be set to Aggressive.
2. If false, unclean.recovery.strategy

...

1. will be set to Balanced.

unclean.leader.election.enable will be marked as deprecated.

Delivery plan

The KIP is a large plan, it can be across multiple quarters. So we have to consider how to deliver the project in phases.

...

Space shortcuts

Child pages

Versions Compared

Old Version 19

New Version Current

Key

Status

Unclean recovery

Public Interfaces

DescribeTopicPartitionsRequest (Coming with ELR)

DescribeTopicResponse

DescribeTopicsResponse

CleanShutdownFile (Coming with ELR)

ElectLeadersRequest (Coming with Unclean Recovery)

GetReplicaLogInfo Request (Coming with Unclean Recovery)

CleanShutdownFile (Coming with ELR)

ElectLeadersRequest (Coming with Unclean Recovery)

GetReplicaLogInfo Response

GetReplicaLogInfo Request (Coming with Unclean Recovery)

kafka-leader-election.sh (Coming with Unclean Recovery)

GetReplicaLogInfo Response

kafka-leader-election.sh (Coming with Unclean Recovery)

Config changes

New Errors

Admin API/Client changes

Metrics

ELR

Clean shutdown

Delivery plan

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 19

New Version Current

Key

Status

Unclean recovery

Public Interfaces

DescribeTopicPartitionsRequest (Coming with ELR)

DescribeTopicResponse

DescribeTopicsResponse

CleanShutdownFile (Coming with ELR)

ElectLeadersRequest (Coming with Unclean Recovery)

GetReplicaLogInfo Request (Coming with Unclean Recovery)

CleanShutdownFile (Coming with ELR)

ElectLeadersRequest (Coming with Unclean Recovery)

GetReplicaLogInfo Response

GetReplicaLogInfo Request (Coming with Unclean Recovery)

kafka-leader-election.sh (Coming with Unclean Recovery)

GetReplicaLogInfo Response

kafka-leader-election.sh (Coming with Unclean Recovery)

Config changes

New Errors

Admin API/Client changes

Metrics

ELR

Clean shutdown

Delivery plan