Status

Current state:   [One of "Under Discussion", "Accepted", "Rejected"]

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

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

JIRA:

Motivation

This KIP is intended to be a complement /alternative to the the unclean recovery design proposal of KIP-966. 

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 replica 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 log 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 set as leader by unclean 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 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. 

...

// Optional values
--recovery-duration-ms <integer>			A broker is required to respond with log information within this duration for it to be considered as a candidate for election.
											Defaults to 30_000 (30 seconds) and represents time in milliseconds.
--recovery-election-duration-ms <integer>   Amount of time to keep attempts <integer>      Number of retries of transient failures allowed for leader election requests.
                                            Defaults to 3. 

// At least 1 of these is required.

--show-replica-info							Print a table showing replica info of targeted partitions. Can be used with any other arguments.

--manual-recovery-output-file <string>		File path of a new file which will be created. The generated file will be capable of being an input to kafka-elect-leaders.sh.
											It will contain topic-partitions and their designated leaders.
                                            {
												"partitions": [
													{"topic": "foo", "partition": 1, "designatedLeader": 0},
	                                        	 	{"topic": "foobar", "partition": 2, "designatedLeader": 1}
												]
                                        	}
											Mutually exclusive with the --automated-recovery flag.

--automated-recovery						Automatically elect leaders with longest apparent logs by attempting designated leader elections. 


// One of these two are required.                    
--path-to-json-file <String>				Path to a JSON file containing a list of topic-partitions to attempt unclean recovery elections on. 
											Mutually exclusive with --all-offline-partitions.
											Example:
											{
												"partitions": [
													{
														topic: "foo",
														partitions: [0, 3, 5]
													},
													{
														topic: "bar",
														partitions: [0, 1, 4]
													}
												]
											}

--all-offline-partitions			Perform unclean recovery on all detected offline partitions.

...

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." }
  ] 
}

...

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": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+", "nullableVersions": "0+",
      "about": "The topic partitions to query the log info for.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID"},
        { "name": "Partitions", "type": "[]int32", "versions": "0+",
          "about": "The partitions of this topic whose leader should be elected." }
      ]}
  ]
}

...

For the tool to work as intended, GetReplicaLogInfo RPC and designated leadership extension to ElectLeaders RPC are needed. 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 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. This behaviour is consistent with the prerequisites for the existing "unclean" leadership elections. 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 "potential" candidate 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. 

kafka-unclean-recovery.sh will attempt a "best-effort" approach and try to complete as many elections as it can within a time limit (or until the operator provides a SIGTERM). It should be expected that not all partitions in the input set will be "recovered" successfully. Possible failure modes are that either ElectLeadersRequest will fail or no successful GetReplicaLogInfoResponse was received in time for a given partition. Since designated leadership elections are idempotent (once a leader is elected, it is not possible to use designated leadership elections again until partition is again offline) operators can confidently run the tool multiple times. A zero exit status will be reported only if all elections for the input set are successful. Any failed elections will be logged to stderr. 

There are some caveats to using a command line tool for this system.

...

A second problem might be that between the tool receiving a GetReplicaLogInfoResponse and sending the ElectLeaderRequest, a an offline ELR replica may become online again and be elected leader. Since designated leader elections can only affect online offline partitions this does not pose a correctness concern, in this case the tool will report that it found a partition already online.

...

An attempt was made to implement this feature within the controller as part of KIP-966. 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. The active controller, due to being the source of truth of cluster metadata, is best positioned to automatically initiate unclean-recoveries as it is immune to a network split between the cluster and operators. However implementation within the controller carries some caveats.

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 chaos within the cluster. Adding this new unclean recoveries 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=true" users would see "unclean.recovery.strategy" as a significant improvement over the current "random" algorithm to justify thisthe complexity. For unclean recovery which is not initiated by the controller, there is not an obvious advantage provided given safety guarantees of designated leadership elections.

...