...

A partition can become in offline if all of the ISR / ELR replicas of a partition are offline. If this occurs, automatic leadership failover will not take place. Unclean recovery is defined as picking a leader for an offline partition from the replica set which was never in the ISR / ELR. Unclean recovery has the potential to lead to data loss. Data loss takes place when the new leader chosen during the recovery election has a lower offset than the high water mark (HWM) of the ISR / ELR replicas. The new leader will always truncate the logs of other replicas. 

Currently, the KRaft quorum controller has the ability to automatically perform unclean recovery elections. This is configured by unclean.leader.election.enable. Users favouring availability over durability have the option to set unclean.leader.election.enable=true which allows the controller to randomly pick a new leader from live replicas every 5 minutes. Users favouring durability over availability will typically configure unclean.election.enable=false. In this case controller will do nothing and the partition will remain offline until an ELR / ISR partition recovers. Unclean recovery elections can be manually triggered using kafka-leader-election.sh using the --election-type=unclean flag. 

In some cases, randomly picking a new leader may not be an ideal outcome. There are no guarantees on offset contents for non-ISR / ELR replicas. A replica more recently added to a replica set may not be as "caught up" as other replicas which have been in the replica set for longer. A pathological example would be a new replica added to the replica set after the partition became offline. This replica would have no data at all and if picked set as leader by unclean leadership election recovery there will be complete truncation of the log. 

Implementation of this KIP will provide a way for operators to "intelligently" recover offline topic partitions during emergencies. It would assist users who run without automated unclean elections ( unclean.leader.election.enable = false ) but may desire a an improved "break glass if needed" option during emergencies. We propose implementing a new command line tool kafka-unclean-recovery.sh which will assist in recovering offline partitions based on log length. 

...

Implementation of kafka-unclean-recovery.sh requires implementation of subset of the RPC changes within KIP-966, specifically a new . These are the GetReplicaLogInfo request as allowing and the modifying ElectLeadersRequest to "designate" a specific replica as the new partition leader. Neither of these two RPCs are yet implemented by the kafka reference implementation. 

kafka-unclean-recovery.sh

...

Designated Leader Elections

Derived from KIP-966.

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": XX,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ElectLeadersRequest",
  "validVersions": "0-3",
  "flexibleVersions": "2+",
  "fields": [
  ...
  { "name": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+", "nullableVersions": "0+",
    "about": "The topic partitions to elect leaders.",
    "fields": [
    ...

// New fields begin. The same level with the Partitions
     { "name": "DesignatedLeaders", "type": "[]int32", "versions": "3+", "nullableVersions": "3+",
       "about": "The designated leaders. The entry should match with the entry in Partitions by the index." },
     },
// New fields end.

  ] },
  { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
    "about": "The time in ms to wait for the election to complete." }
  ] 
}

...

Originally defined in KIP-966. 

Request

ACL: CLUSTER_ACTION, DESCRIBE (on a topic)

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":XX,
  "type": "request",
  "listeners": ["broker"],
  "name": "GetReplicaLogInfoRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "BrokerIdTopicPartitions", "type": "int32[]TopicPartitions", "versions": "0+", "entityTypenullableVersions": "brokerId0+", 
        "about": "The ID oftopic partitions to query the log info brokerfor." },
    { "name": "TopicPartitionsfields", "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", "versions": "0+",
          "about": "The partitions of this topic whose leader should be elected." },
      ]}
  ] 
}

Response

Differences from KIP-966:

1. Addition of HasMoreData field which is true if there were more than 1000 partitions. In that case we only return the first 1000 and set HasMoreData = true.
2. Removal of LastWrittenLeaderEpoch which refers to the previous Raft Leader epoch. This is a safety property is not useful for a command line tool.  

{
  "apiKey":XX,
  "type": "response",
  "name": "GetReplicaLogInfoResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "BrokerEpochThrottleTimeMs", "type": "int64int32", "versions": "0+",
      "about": "The duration in epochmilliseconds for which the broker request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "TopicPartitionLogInfoListBrokerEpoch", "type": "[]TopicPartitionLogInfoint64", "versions": "0+",
 
     "about": "The listepoch of the log infobroker." },
    { "fieldsname": [
      { "name": "TopicId"HasMoreData", "type": "uuidbool", "versions": "0+", "ignorable": true,
      "about": "The unique topic IDTrue if response does not include all the topic partitions requested. Only the first 1000 topic partitions are returned."},
      { "name": "PartitionLogInfoTopicPartitionLogInfoList", "type": "[]PartitionLogInfoTopicPartitionLogInfo", "versions": "0+",
      "about": "The loglist info of athe partition log info.", 
        "fields": [
        { "name": "PartitionTopicId", "type": "int32uuid", "versions": "0+", "about": "The idunique fortopic the partitionID." },
        { "name": "LastWrittenLeaderEpochPartitionLogInfo", "type": "int32[]PartitionLogInfo", "versions": "0+", "about": "The last written leader epoch in the log." }, log info of a partition.",
          "fields": [
            { "name": "CurrentLeaderEpochPartition", "type": "int32", "versions": "0+", "about": "The currentid leader epoch for the partition." from},
 the broker point of view." },
      { "name": "LogEndOffsetPartitionLeaderEpoch", "type": "int64int32", "versions": "0+", "about": "The logcurrent endleader offsetepoch for the partition from the broker point of view." },
            { "name": "ErrorCodeLogEndOffset", "type": "int16int64", "versions": "0+", "about": "The resultlog error,end oroffset zerofor if there was no errorthe partition." },
      { "name": "ErrorMessage", "type     { "name": "stringErrorCode", "versionstype": "0+int16", "nullableVersionsversions": "0+", "about": "The result messageerror, or nullzero if there was no error."}
        ]},
      ]}
    ] 
}

Proposed Changes

For the tool to work as intended, GetReplicaLogInfo RPC and designated leadership extension to ElectLeaders RPC are needed. Designated leadership elections While it is not explicitly specified in KIP-966, designated leadership must have specific safety properties for the tool to operate safely. Designated leader elections should only be allowed for partitions if the partition is offline (IE has no leader). A broker may only be designated a leader of a partition if it is in the replica set, is unfenced and has an online replica. GetReplicaLogInfo requests ask the broker for the length of its replica of a given partition. Note GetReplicaLogInfo in this KIP is slightly different from the one specified in KIP-966.  

kafka-unclean-recovery.sh will accept an input set of topic-partitions to attempt to recover. It will send GetReplicaLogInfoRequests to each of the brokers to figure out their log lengths. It will keep retrying these requests until it hits a configurable deadline. From the GetReplicaLogInfoResponse's it has received it will attempt to figure out the best leader for each topic partition by selecting the replica with the highest epoch and then longest log length. After the best leader has been selected, the tool will either send leader election requests to the controller or output a file which can be used as an input to the kafka-elect-leader.sh tool with the intention that the operator can review/modify the requests and/or send them manually. GetReplicaLogInfo RPC request can be amoritized by broker but this is an optimization and not required for correctness. 

...

This approach is purely additive and is not incompatible with any existing RPCs.

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

The core correctness concern of the tool can be tested A system test could be setup to run in the following way:

1. Spins up a test cluster unclean.leader.election=false and create a partition. Produce to it.
2. Terminate all brokers in the ISR with unclean shutdowns (`kill -9`) and ensure that no the entire ISR is createdoffline. 
3. Assert that the test partition is offline. 
4. Record the replica set.
5. Get kafka-unclean-recovery.sh to perform a designated leadership election using --automated-recovery flag. 
6. Assert that the test partition is online and is using a broker from the non-ISR replica set.

Rejected Rejected Alternatives

An attempt was made to implement this feature within the controller as part of KIP-966. Implementing this functionality within the controller does not provide KIP-966 proposed adding a new "unclean.recovery.strategy" configuration to the controller which contained two new "automated" election strategies - "Balanced" and "Aggressive". Both new strategies would allow recovery by discovering the replicas with longest logs. The new configuration would operate in a similar fashion to "unclean.election.enable" - IE it would be initiated by the controller every 5 minutes. Reliable automatic strategies are the chief advantage of implementing unclean recovery within the controller. 

In normal operating cases, failover to another controller will happen automatically. Unclean recovery is intended to be used during emergencies where it is possible that the quorum may be partially degraded or the controller is extremely busy keeping track of activity in the cluster. Adding this to the controller also adds complexity to the logic and operation of a sensitive aspect of cluster operation (leadership election) during emergencies. It is unclear whether users "unclean.election.enable" would see "unclean.recovery.strategy" as a significant improvement to justify this. For unclean recovery which is not initiated by the controller, there is not an obvious advantage provided given safety guarantees of designated leadership elections.

The KRaft controller operates in a "pull" model - Raft observers send fetch requests to the controller to receive up to date information. The controller handling unclean recovery would have required management of many GetReplicaLogInfo requests in a "push" model which could raise concerns about memory usage and could affect quota calculations. Further, the ElectLeaderRequest API is intended to be "synchronous" with respect to replication through KRaft. Allowing "longest log" recovery elections would have subtly changed this API contract as it now needs to wait for a more complex network requests sent out to brokers in comparison to just waiting for "normal" KRaft replication. 

KIP-966 proposed adding a new "unclean.recovery.strategy" configuration to the controller. The the new "Balanced" and "Aggressive" election configurations add a larger API surface - with 3 additional election types becoming possible. The new configurations add ambiguity to unclean.election.enable configuration which would need to live with the new unclean.recovery.strategy. The command line tool approach only requires adding designated elections and moves the question of how an election should work outside of the controller and the impetus to add additional recovery strategies. In theory, users can implement their own specific strategy using GetReplicaLogInfo and designated leadership elections - there is no need for the controller to pick or have a specific strategy in mind. 

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.the kafka-unclean-recovery.sh tool fails catastrophically then this has no effect on the liveness of the controller.