THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
| Code Block | ||
|---|---|---|
| ||
git diff upstream/trunk raft/src/main/resources/common/message/QuorumStateData.json
diff --git a/raft/src/main/resources/common/message/QuorumStateData.json b/raft/src/main/resources/common/message/QuorumStateData.json
index d71a32c75d..34881f05ff 100644
--- a/raft/src/main/resources/common/message/QuorumStateData.json
+++ b/raft/src/main/resources/common/message/QuorumStateData.json
@@ -16,19 +16,25 @@
{
"type": "data",
"name": "QuorumStateData",
- "validVersions": "0",
+ "validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
- {"name": "ClusterId", "type": "string", "versions": "0+"},
- {"name": "LeaderId", "type": "int32", "versions": "0+", "default": "-1"},
- {"name": "LeaderEpoch", "type": "int32", "versions": "0+", "default": "-1"},
- {"name": "VotedId", "type": "int32", "versions": "0+", "default": "-1"},
- {"name": "AppliedOffset", "type": "int64", "versions": "0+"},
- {"name": "CurrentVoters", "type": "[]Voter", "versions": "0+", "nullableVersions": "0+"}
+ { "name": "ClusterId", "type": "string", "versions": "0" },
+ { "name": "LeaderId", "type": "int32", "versions": "0+", "default": "-1" },
+ { "name": "LeaderEpoch", "type": "int32", "versions": "0+", "default": "-1" },
+ { "name": "VotedId", "type": "int32", "versions": "0+", "default": "-1" },
+ { "name": "VotedUuid", "type": "uuid", "versions": "1+" },
+ { "name": "AppliedOffset", "type": "int64", "versions": "0" },
+ { "name": "AppliedRecord", "type": "AppliedRecord", "versions": "1+", "fields": [
+ { "name": "EndOffset", "type": "int64", "versions": "1+" },
+ { "name": "Epoch", "type": "int32", "versions": "1+" }
+ ]},
+ { "name": "CurrentVoters", "type": "[]Voter", "versions": "0+", "nullableVersions": "0+" }
],
"commonStructs": [
{ "name": "Voter", "versions": "0+", "fields": [
- {"name": "VoterId", "type": "int32", "versions": "0+"}
+ { "name": "VoterId", "type": "int32", "versions": "0+" },
+ { "name": "VoterUuid", "type": "uuid", "versions": "1+" }
]}
]
} |
RPCs
AddVoter
Request
This RPC can be sent by an administrative client to add a voter to the set of voters. This RPC can be sent to a broker or controller. The , when sent to a broker, the broker will forward the request to the controller.
Handling
When the leader receives an AddVoter request it will do the following:
- Wait for the fetch offset of the replica (ID, UUID) to catch up to the log end offset of the leader.
- Wait for until there are no uncommitted add or remove voter records.
- Wait for the LeaderChangeMessage control record from the current epoch to get committed.
- Send an ApiVersions RPC to the first listener to discover the supported
kraft.versionof the new voter. - Check that the new voter supports the current
kraft.version. - Append the AddVoterRecord to the log.
- The KRaft internal listener will read this record from the log and add the voter to the voter set.
- Wait for the AddVoterRecord to commit using the majority of new voter set.
- Send the AddVoter response to the client.
In 1., the leader needs to wait for the replica to catch up because when the AddVoterRecord is appended to the log, the set of voter changes. If the new voter is too far behind then it can take some time for it to reach the HWM. During this time the leader cannot commit data and the quorum will be unavailable from the perspective of the state machine. We mitigate this by waiting for the new replica to catch up before adding it to the set of voters.
In 4., the new replica will be part of the quorum so the leader will start sending BeginQuorumEpoch requests to this replica. It is possible that the new replica has not yet replicated and applied this AddVoterRecord so it doesn't know that it is a voter for this topic partition. The new replica will accept the BeginQuorumEpoch RPC even if it is doesn't believe it is a member of the voter set.
The replica will return the following errors:
- NOT_LEADER_FOR_PARTITION - when the request is sent to a replica that is not the leader.
- VOTER_ALREADY_ADDED - when the request contains a replica ID and UUID that is already in the committed voter set.
- UNSUPPORTED_VERSION - when the
kraft.versionis not greater than 1. - INVALID_REQUEST - when the new voter doesn't support the current
kraft.version.
Request
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 75,
"type": "request",
"listeners": ["controller", "broker"],
"name": "AddVoterRequest",
"validVersions": "0",
"flexibleVersions | ||
| Code Block | ||
| ||
{ "apiKey": "TBD", "type": "request", "listeners": ["controller", "broker"], "name": "AddVoterRequest", "validVersions": "0", "flexibleVersions": "0+", "fields": [ { "name": "ClusterId", "type": "string", "versions": "0+" } { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [ { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName", "about": "The name of the topic." }, { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [ { "name": "Index", "type": "int32", "versions": "0+", "fields": [ { "aboutname": "The partition index."ClusterId", "type": "string", "versions": "0+" }, { "name": "VoterIdTopicName", "type": "int32string", "versions": "0+", "entityType": "topicName", "about": "The IDname of the voter getting added to the topic partition." }, { "name": "VoterUuidTopicId", "type": "uuid", "versions": "0+", "about": "The replica generated UUID of the replica getting added as a voter to the topic partition.unique topic ID" }, { "name": "EndPointsPartition", "type": "[]Endpointint32", "versions": "0+", "about": "The endpointpartition that can be used to communicate with the leader", "fields": [ index" }, { "name": "NameVoterId", "type": "stringint32", "versions": "0+", "mapKey": true, "about": "The nameID of the voter endpoint." }, getting added to the topic partition" }, { "name": "HostVoterUuid", "type": "stringuuid", "versions": "0+", "about": "The hostname." }, directory id of the voter getting added to the topic partition" }, { "name": "PortListeners", "type": "uint16[]Listener", "versions": "0+", "about": "The endpoints that can be "about": "The port." }, used to communicate with the voter", "fields": [ { "name": "SecurityProtocolName", "type": "int16string", "versions": "0+", "mapKey": true, "about": "The securityname protocol." } ]} ]}of the endpoint" }, ]} ] } |
Response
| Code Block | ||
|---|---|---|
| ||
{
"apiKeyname": "TBDHost",
"type": "responsestring",
"nameversions": "AddVoterResponse0+",
"validVersions": "0",
"flexibleVersionsabout": "0+"The hostname" },
"fields": [
{ "name": "ErrorCodePort", "type": "int16uint16", "versions": "0+",
"about": "The top level error code.port" },
{ "name": "TopicsSecurityProtocol", "type": "[]TopicDataint16", "versions": "0+",
"fieldsabout": [ "The security protocol" }
]}
]
} |
Response
| Code Block | ||
|---|---|---|
| ||
{
"nameapiKey": "TopicName"75,
"type": "stringresponse",
"versionsname": "0+AddVoterResponse",
"entityTypevalidVersions": "topicName0",
"about"flexibleVersions": "The name of the topic." },"0+",
"fields": [
{ "name": "PartitionsErrorCode", "type": "[]PartitionDataint16", "versions": "0+", "fields": [
{ "name": "Index", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no fetch error." },
{ "name": "CurrentLeader", "type": "LeaderIdAndEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
{ "name": "LeaderId", "type": "int32", "versions": "120+", "default": "-1", "entityType" : "brokerId",
"about": "The IDreplica id of the current leader or -1 if the leader is unknown." },
{ "name": "LeaderEpoch", "type": "int32", "versions": "120+", "default": "-1",
"about": "The latest known leader epoch." },
]},
{ "name": "EndPointNodeEndpoint", "type": "EndpointNodeEndpoint", "versions": "0+",
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "0+",
"about": "The hostname." },
"taggedVersions": "0+", "tag": 1,
{ "name": "Port", "type": "uint16", "versions": "0+",
"about": "The port." }
]}
]}
]}
]}
]
} |
Handling
When the leader receives a AddVoter request it will do the following:
- Wait for the fetch offset of the replica (ID, UUID) to catch up to the log end offset of the leader.
- Wait for until there are no uncommitted add or remove voter records.
- Wait for the LeaderChangeMessage control record from the current epoch to get committed.
- Append the AddVoterRecord to the log.
- The KRaft internal listener will read this record from the log and add the voter to the voter set.
- Wait for the AddVoterRecord to commit using the majority of new configuration.
- Send the AddVoter response to the client.
In 1., the leader needs to wait for the replica to catch up because when the AddVoterRecord is appended to the log the set of voter changes. If the added voter is far behind then it can take some time for it to reach the HWM. During this time the leader cannot commit data and the quorum will be unavailable from the perspective of the state machine. We mitigate this by waiting for the new replica to catch up before adding it to the set of voters.
In 4., the new replica will be part of the quorum so the leader will start sending BeginQuorumEpoch requests to this replica. It is possible that the new replica has not yet replicated and applied this AddVoterRecord so it doesn't know that it is a voter for this topic partition. The new replica will fail this RPC until it discovers that it is in the voter set. The leader will continue to retry until the RPC succeeds.
The replica will return the following errors:
- NOT_LEADER_FOR_PARTITION - when the request is sent to a replica that is not the leader.
- VOTER_ALREADY_ADDED - when the request contains a replic ID and UUID that is already in the committed voter set.
- INVALID_REQUEST - when the request contains a replica ID and UUID that is not an observer.
- INVALID_REQUEST - when the request contains endpoints but the quorum was configured to use controller.quorum.voters.
- INVALID_REQUEST - when the request contains a replica ID that is not enumerated in the controller.quorum.voters configuration.
...
about": "Endpoint for current leader of the topic partition", "fields": [
{ "name": "Host", "type": "string", "versions": "0+", "about": "The node's hostname" },
{ "name": "Port", "type": "int32", "versions": "0+", "about": "The node's port" }
]}
]
} |
RemoveVoter
Request
This RPC can be sent by an administrative client to remove a voter from the set of voters. This RPC can be sent to a broker or controller. The broker will forward the request to the controller.
...
The broker will handle this request by forwarding the request to the active controller.
UpdateVoter
TODO: fill this out
Request
Response
Handling
Fetch
Request
Version 14 adds the field ReplicaUuid to the FetchPartition. This field is populated with the replica generated UUID. If the ReplicaUuid and the ReplicaId fields are populated, the topic partition leader can assume that the replica supports becoming a voter.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 1,
"type": "request",
"listeners": ["zkBroker", "broker", "controller"],
"name": "FetchRequest",
"validVersions": "0-14",
"flexibleVersions": "12+",
"fields": [
{ "name": "ClusterId", "type": "string", "versions": "12+", "nullableVersions": "12+", "default": "null", "taggedVersions": "12+", "tag": 0, "ignorable": true,
"about": "The clusterId if known. This is used to validate metadata fetches prior to broker registration." },
{ "name": "ReplicaId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The replica ID of the follower, of -1 if this request is from a consumer." },
...
{ "name": "Topics", "type": "[]FetchTopic", "versions": "0+",
"about": "The topics to fetch.", "fields": [
{ "name": "Topic", "type": "string", "versions": "0-12", "entityType": "topicName", "ignorable": true,
"about": "The name of the topic to fetch." },
{ "name": "TopicId", "type": "uuid", "versions": "13+", "ignorable": true,
"about": "The unique topic ID"},
{ "name": "Partitions", "type": "[]FetchPartition", "versions": "0+",
"about": "The partitions to fetch.", "fields": [
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ReplicaUuid", "type": "uuid", "versions": "14+", "nullableVersions": "14+", "default": "null",
"about": "The replica generated UUID. null otherwise." },
...
]}
]}
]
} |
Response
Version 14 rename LeaderIdAndEpoch to CurrentLeader and adds Endpoint to CurrentLeader.
| Code Block |
|---|
{
"apiKey": 1,
"type": "response",
"name": "FetchResponse",
"validVersions": "0-14",
"flexibleVersions": "12+",
"fields": [
...
{ "name": "Responses", "type": "[]FetchableTopicResponse", "versions": "0+",
"about": "The response topics.", "fields": [
{ "name": "Topic", "type": "string", "versions": "0-12", "ignorable": true, "entityType": "topicName",
"about": "The topic name." },
{ "name": "TopicId", "type": "uuid", "versions": "13+", "ignorable": true, "about": "The unique topic ID"},
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
"about": "The topic partitions.", "fields": [
...
{ "name": "CurrentLeader", "type": "CurrentLeader",
"versions": "12+", "taggedVersions": "12+", "tag": 1, "fields": [
{ "name": "LeaderId", "type": "int32", "versions": "12+", "default": "-1", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "12+", "default": "-1",
"about": "The latest known leader epoch"},
{ "name": "EndPoint", "type": "Endpoint", "versions": "14+",
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "14+",
"about": "The hostname." },
{ "name": "Port", "type": "uint16", "versions": "14+",
"about": "The port." }
]}
]},
...
]}
]}
]
} |
Handling
Replica that support becoming voters will send both the replica ID and UUID in the Fetch request. The leader will assume that replicas that report both fields are voters or are able to become voters.
...
When removing the leader from the voter set, it will remain the leader for that epoch until the RemoveVoterRecord gets committed. This means that the leader needs to allow replicas (voters and observers) to fetch from the leader even if it is not part of the voter set. This also means that if the leader is not part of the voter set it should not include itself when computing the committed offset (also known as the high-watermark).
FetchSnapshot
Request
Version 1 adds the field ReplicaUuid to PartitionSnapshot. If the ReplicaUuid and the ReplicaId fields are populated, the topic partition leader can assume that the replica supports becoming a voter.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 59,
"type": "request",
"listeners": ["controller"],
"name": "FetchSnapshotRequest",
"validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
{ "name": "ClusterId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", "taggedVersions": "0+", "tag": 0,
"about": "The clusterId if known, this is used to validate metadata fetches prior to broker
registration" },
{ "name": "ReplicaId", "type": "int32", "versions": "0+", "default": "-1", "entityType": "brokerId",
"about": "The replica ID of the follower" },
{ "name": "MaxBytes", "type": "int32", "versions": "0+", "default": "0x7fffffff",
"about": "The maximum bytes to fetch from all of the snapshots" },
{ "name": "Topics", "type": "[]TopicSnapshot", "versions": "0+",
"about": "The topics to fetch", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The name of the topic to fetch" },
{ "name": "Partitions", "type": "[]PartitionSnapshot", "versions": "0+",
"about": "The partitions to fetch", "fields": [
{ "name": "Partition", "type": "int32", "versions": "0+",
"about": "The partition index" },
{ "name": "ReplicaUuid", "type": "uuid", "versions": "1+", "default": "null",
"about": "The replica UUID of the follower" },
{ "name": "CurrentLeaderEpoch", "type": "int32", "versions": "0+",
"about": "The current leader epoch of the partition, -1 for unknown leader epoch" },
{ "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
"about": "The snapshot endOffset and epoch to fetch", "fields": [
{ "name": "EndOffset", "type": "int64", "versions": "0+" },
{ "name": "Epoch", "type": "int32", "versions": "0+" }
]},
{ "name": "Position", "type": "int64", "versions": "0+",
"about": "The byte position within the snapshot to start fetching from" }
]}
]}
]
} |
Response
Version 1 renames LeaderIdAndEpoch to CurrentLeader and adds Endpoint to CurrentLeader.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 59,
"type": "response",
"name": "FetchSnapshotResponse",
"validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": 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": "ErrorCode", "type": "int16", "versions": "0+", "ignorable": false,
"about": "The top level response error code." },
{ "name": "Topics", "type": "[]TopicSnapshot", "versions": "0+",
"about": "The topics to fetch.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The name of the topic to fetch." },
{ "name": "Partitions", "type": "[]PartitionSnapshot", "versions": "0+",
"about": "The partitions to fetch.", "fields": [
{ "name": "Index", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no fetch error." },
{ "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
"about": "The snapshot endOffset and epoch fetched",
"fields": [
{ "name": "EndOffset", "type": "int64", "versions": "0+" },
{ "name": "Epoch", "type": "int32", "versions": "0+" }
]},
{ "name": "CurrentLeader", "type": "CurrentLeader",
"versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
{ "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The latest known leader epoch"},
{ "name": "EndPoint", "type": "Endpoint", "versions": "1+",
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "1+",
"about": "The hostname." },
{ "name": "Port", "type": "uint16", "versions": "1+",
"about": "The port." }
]},
{ "name": "Size", "type": "int64", "versions": "0+",
"about": "The total size of the snapshot." },
{ "name": "Position", "type": "int64", "versions": "0+",
"about": "The starting byte position within the snapshot included in the Bytes field." },
{ "name": "UnalignedRecords", "type": "records", "versions": "0+",
"about": "Snapshot data in records format which may not be aligned on an offset boundary" }
]}
]}
]
}
|
Handling
Similar to Fetch, replica that support becoming voters will send both the replica ID and UUID in the FetchSnapshot request. The leader will assume that replicas that report both fields are voters or are able to become voters.
...
When removing the leader from the voter set, it will remain the leader for that epoch until the RemoveVoterRecord gets committed. This means that the leader needs to allow replicas (voters and observers) to fetch snapshots from the leader even if it is not part of the voter set.
Vote
Request
Changes:
- Candidate Id was moved out of the topic partition maps
- Candidate Uuid was added to the PartitionData
- VoterId was added to the top level
- VoterUuId was added to PartitionData
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 52,
"type": "request",
"listeners": ["controller"],
"name": "VoteRequest",
"validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
{ "name": "ClusterId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null" },
{ "name": "CandidateId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The ID of the voter sending the request" },
{ "name": "VoterId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The ID of the replica receiving the request to vote." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData",
"versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "CandidateEpoch", "type": "int32", "versions": "0+",
"about": "The bumped epoch of the candidate sending the request"},
{ "name": "CandidateId", "type": "int32", "versions": "0", "entityType": "brokerId",
"about": "The ID of the voter sending the request"},
{ "name": "CandidateUuid", "type": "uuid", "versions": "1+",
"about": "The candidate generated UUID, null otherwise" },
{ "name": "VoterUuid", "type": "uuid", "versions": "1+", "nullableVersions": "1+", "default": "null" }
"about": "The replica generated UUID of the replica receiving the request to vote, null otherwise" },
{ "name": "LastOffsetEpoch", "type": "int32", "versions": "0+",
"about": "The epoch of the last record written to the metadata log"},
{ "name": "LastOffset", "type": "int64", "versions": "0+",
"about": "The offset of the last record written to the metadata log"}
]}
]}
]
} |
Response
Version 1 addes the tagged field LeaderEndpoint to PartitionData.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 52,
"type": "response",
"name": "VoteResponse",
"validVersions": "0-1",
"flexibleVersions": "0+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top level error code."},
{ "name": "Topics", "type": "[]TopicData",
"versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData",
"versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+"},
{ "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The latest known leader epoch"},
{ "name": "VoteGranted", "type": "bool", "versions": "0+",
"about": "True if the vote was granted and false otherwise"},
{ "name": "VoterUuid", "type": "bool", "versions": "1+",
"about": "The replica generated uuid for the replica casting a vote." },
{ "name": "LeaderEndPoint", "type": "Endpoint", "versions": "1+", "taggedVersions": "+1", "tag": 0,
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "1+",
"about": "The hostname." },
{ "name": "Port", "type": "uint16", "versions": "1+",
"about": "The port." }
]}
]}
]}
]
} |
Handling
The handling of the Vote request will from that described in KIP-595 in that all replicas are allowed to vote event if they are not voters for the quorum. The voter will persist in the quorum state both the candidate ID and UUID so that it only votes for at most one candidate for a given epoch.
...
TODO: Talk about the leader sending this request as part of quorum check. This is needed because the voter could come back with a new replica uuid. In that case voter can only discover the leader using Fetch unless the leader sends the begin quorum epoch request.
TODO: The receiver of the RPC always applies the request even if it is not a member of the voter set.
Request
- LeaderId was moved out of the topic partition maps
- VoterId was added to the top level
- VoterUuId was added to PartitionData
- Allow tagged fields for version greater than or equal to 1.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 53,
"type": "request",
"listeners": ["controller"],
"name": "BeginQuorumEpochRequest",
"validVersions": "0-1",
"flexibleVersions": "1+",
"fields": [
{ "name": "ClusterId", "type": "string", "versions": "0+",
"nullableVersions": "0+", "default": "null"},
{ "name": "LeaderId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The ID of the newly elected leader"},
{ "name": "VoterId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The voter ID of the receiving replica." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "VoterUuid", "type": "uuid", "versions": "1+",
"about": "The replica UUID of the receiving replica." },
{ "name": "LeaderId", "type": "int32", "versions": "0", "entityType": "brokerId",
"about": "The ID of the newly elected leader"},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The epoch of the newly elected leader"}
]}
]}
]
} |
Response
Version 1 is a flexible version and add the tagged field LeaderEndpoint to PartitionData.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 53,
"type": "response",
"name": "BeginQuorumEpochResponse",
"validVersions": "0-1",
"flexibleVersions": "1+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top level error code."},
{ "name": "Topics", "type": "[]TopicData",
"versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData",
"versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+"},
{ "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The latest known leader epoch"},
{ "name": "LeaderEndPoint", "type": "Endpoint", "versions": "1+", "taggedVersions": "+1", "tag": 0,
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "1+",
"about": "The hostname." },
{ "name": "Port", "type": "uint16", "versions": "1+",
"about": "The port." }
]}
]}
]}
]
} |
Handling
This request will be handle as described in KIP-595 with the following additional errors:
- INVALID_REQUEST - when the voter ID and UUID doesn't match.
- NOT_VOTER - when the the replica is not a voter in the topic partition. This error should be retry by the leader of the topic partition.
EndQuorumEpoch
Request
- LeaderId was moved out of the topic partition maps
- VoterId was added to the request
- VoterUuid was added to the Partitions
- ReplicaUuid was added to PreferredSuccessors
- Allow tagged fields for versions greater than or equal to 1.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 54,
"type": "request",
"listeners": ["controller"],
"name": "EndQuorumEpochRequest",
"validVersions": "0-1",
"flexibleVersions": "1+",
"fields": [
{ "name": "ClusterId", "type": "string", "versions": "0+",
"nullableVersions": "0+", "default": "null"},
{ "name": "LeaderId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The current leader ID that is resigning." },
{ "name": "VoterId", "type": "int32", "versions": "1+", "entityType": "brokerId",
"about": "The voter ID of the receiving replica." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "VoterUuid", "type": "uuid", "versions": "1+",
"about": "The replica UUID of the receiving replica." },
{ "name": "LeaderId", "type": "int32", "versions": "0", "entityType": "brokerId",
"about": "The current leader ID that is resigning"},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The current epoch"},
{ "name": "PreferredSuccessors", "type": "[]ReplicaInfo", "versions": "0+",
"about": "A sorted list of preferred successors to start the election", "fields": [
{ "name": "ReplicaId", "type": "int32", "versions": "0+", "entityType": "brokerId" },
{ "name": "ReplicaUuid", "type": "uuid", "versions": "1+" }
]}
]}
]}
]
} |
Response
Version 1 is a flexible version and add the tagged field LeaderEndpoint to PartitionData.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 54,
"type": "response",
"name": "EndQuorumEpochResponse",
"validVersions": "0-1",
"flexibleVersions": "1+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top level error code."},
{ "name": "Topics", "type": "[]TopicData",
"versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData",
"versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+"},
{ "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The latest known leader epoch"},
{ "name": "LeaderEndPoint", "type": "Endpoint", "versions": "1+", "taggedVersions": "+1", "tag": 0,
"about": "The endpoint that can be used to communicate with the leader", "fields": [
{ "name": "Host", "type": "string", "versions": "1+",
"about": "The hostname." },
{ "name": "Port", "type": "uint16", "versions": "1+",
"about": "The port." }
]}
]}
]}
]
} |
Handling
This request will be handle as described in KIP-595 with the following additional errors:
...
The version of the request is increase and the fields remain unchanged.
Response
- Add ReplicaUuid to ReplicaState
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 55,
"type": "response",
"name": "DescribeQuorumResponse",
"validVersions": "0-2",
"flexibleVersions": "0+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top level error code."},
{ "name": "Topics", "type": "[]TopicData",
"versions": "0+", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData",
"versions": "0+", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+"},
{ "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "0+",
"about": "The latest known leader epoch"},
{ "name": "HighWatermark", "type": "int64", "versions": "0+"},
{ "name": "CurrentVoters", "type": "[]ReplicaState", "versions": "0+" },
{ "name": "Observers", "type": "[]ReplicaState", "versions": "0+" }
]}
]}
],
"commonStructs": [
{ "name": "ReplicaState", "versions": "0+", "fields": [
{ "name": "ReplicaId", "type": "int32", "versions": "0+", "entityType": "brokerId" },
{ "name": "ReplicaUuid", "type": "uuid", "versions": "2+" },
{ "name": "LogEndOffset", "type": "int64", "versions": "0+",
"about": "The last known log end offset of the follower or -1 if it is unknown"},
{ "name": "LastFetchTimestamp", "type": "int64", "versions": "1+", "ignorable": true, "default": -1,
"about": "The last known leader wall clock time time when a follower fetched from the leader. This is reported as -1 both for the current leader or if it is unknown for a voter" },
{ "name": "LastCaughtUpTimestamp", "type": "int64", "versions": "1+", "ignorable": true, "default": -1,
"about": "The leader wall clock append time of the offset for which the follower made the most recent fetch request. This is reported as the current time for the leader and -1 if unknown for a voter" }
]}
]
} |
Handling
The handling of the request is the same as that described in KIP-595 with just the additional fields. Clients handling the response can assume that if a ReplicaState include both the ID and the UUID that replica can become a voter if is enumerated in the Observers field.
...