THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
In the ProduceResponse & FetchResponse, new leader’s connection parameters (host & port) will also be supplied along with LeaderId & LeaderEpoch, upon returning errors NOT_LEADER_OR_FOLLOWER and FENCED_LEADER_EPOCH . This will be useful in situations when new leader's connection parameters are not part of client's metadata cache or stale. This can happen in a scenario where an existing broker restarts, or a new broker is added to the cluster. For instance when existing broker restarts, any subsequent metadata refresh on the client, will remove the connection parameters from its cache while broker is shutdown. After broker is restarted, and this broker becomes the new leader, client will require connection parameters along with LeaderId and LeaderEpoch to connect to this new leader in a subsequent retry of the failed produce or fetch request. To this effect, LeaderHost and LeaderPort are included in the responses. Additionally LeaderRack host & port is returned for all such brokers. Additionally rack is returned to be consistent with the broker information returned in MetadataResponse. This will simply override the previously stored broker node information in the metadata cached locally.
...
FetchResponse already contains CurrentLeader introduced in KIP-595, which will be used to propagate LeaderId & LeaderEpoch. Additionally CurrentLeaderBroker NodeEndpoints is introduced to propagate LeaderHosthost, LeaderPort port & LeaderRach rack for the leader. CurrentLeaderBroker is current-leaders enumerated across all PartitionDatas. NodeEndpoints is a tagged field, which is a minor optimisation of saving bytes on the network, as it won’t be set always. This doesn’t require a version bump.
...
| Code Block |
|---|
{
"apiKey": 1,
"type": "response",
"name": "FetchResponse",
"validVersions": "0-15",
"flexibleVersions": "12+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "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": "7+", "ignorable": true,
"about": "The top level response error code." },
{ "name": "SessionId", "type": "int32", "versions": "7+", "default": "0", "ignorable": false,
"about": "The fetch session ID, or 0 if this is not part of a fetch session." },
{ "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": "PartitionIndex", "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": "HighWatermark", "type": "int64", "versions": "0+",
"about": "The current high water mark." },
{ "name": "LastStableOffset", "type": "int64", "versions": "4+", "default": "-1", "ignorable": true,
"about": "The last stable offset (or LSO) of the partition. This is the last offset such that the state of all transactional records prior to this offset have been decided (ABORTED or COMMITTED)" },
{ "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true,
"about": "The current log start offset." },
{ "name": "DivergingEpoch", "type": "EpochEndOffset", "versions": "12+", "taggedVersions": "12+", "tag": 0,
"about": "In case divergence is detected based on the `LastFetchedEpoch` and `FetchOffset` in the request, this field indicates the largest epoch and its end offset such that subsequent records are known to diverge",
"fields": [
{ "name": "Epoch", "type": "int32", "versions": "12+", "default": "-1" },
{ "name": "EndOffset", "type": "int64", "versions": "12+", "default": "-1" }
]},
{ "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
"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"}
]},
// ---------- Start new field ----------
{ "name": "CurrentLeaderBrokerSnapshotId", "type": "BrokerInfoSnapshotId",
"versions": "1512+", "taggedVersions": "1512+", "tag": 2,
"about": "CurrentLeader's broker informationIn the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request.",
"fields": [
{ "name": "LeaderHostEndOffset", "type": "stringint64", "versions": "150+",
"aboutdefault": "The leader hostname.-1" },
{ "name": "LeaderPortEpoch", "type": "int32", "versions": "150+",
"default": "-1" }
"about": "The leader port." ]},
{ "name": "LeaderRackAbortedTransactions", "type": "string[]AbortedTransaction", "versions": "154+", "ignorablenullableVersions": true"4+", "defaultignorable": "null"true,
"about": "The rack of the leader, or null if it has not been assigned to a rack." }
]},
aborted transactions.", "fields": [
{ "name": "ProducerId", "type": "int64", "versions": "4+", "entityType": "producerId",
// ---------- End new field ----------
{ "name": "SnapshotId", "type": "SnapshotId",
"about": "The producer id associated with the aborted transaction." },
{ "versionsname": "12+FirstOffset", "taggedVersionstype": "12+int64", "tagversions": 2"4+",
"about": "In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used The first offset in the FetchSnapshotaborted requesttransaction.", }
"fields": [
]},
{ "name": "EndOffsetPreferredReadReplica", "type": "int64int32", "versions": "011+", "default": "-1" },
"ignorable": false, "entityType": "brokerId",
{ "name": "Epoch", "type"about": "int32", "versions": "0+", "default": "-1" }
]The preferred read replica for the consumer to use on its next fetch request"},
{ "name": "AbortedTransactionsRecords", "type": "[]AbortedTransactionrecords", "versions": "40+", "nullableVersions": "40+", "ignorableabout": true,
"The record data."}
"about": "The aborted transactions.", "fields": [
]}
]},
// ---------- Start new field ----------
{ "name": "ProducerIdNodeEndpoints", "type": "int64[]NodeEndpoint", "versions": "412+", "entityTypetaggedVersions": "producerId12+",
"tag": 0,
""about": "TheEndpoints producerfor idall associatedcurrent-leaders withenumerated thein aborted transactionPartitionData." },
"fields": [
{ "name": "FirstOffsetNodeId", "type": "int64int32", "versions": "412+",
"mapKey": true, "entityType": "brokerId", "about": "The firstID offsetof in the abortedassociated transactionnode." }
]},
{ "name": "PreferredReadReplicaHost", "type": "int32string", "versions": "11+", "default": "-1", "ignorable": false, "entityType": "brokerId",
12+",
"about": "The preferred read replica for the consumer to use on its next fetch request"node's hostname." },
{ "name": "RecordsPort", "type": "recordsint32", "versions": "012+",
"nullableVersions": "0+", " "about": "The node's port." },
{ "name": "Rack", "type": "string", "versions": "12+", "nullableVersions": "12+", "ignorable": true, "default": "null",
"about": "The record data." rack of the node, or null if it has not been assigned to a rack." }
]}
// ---------- End ]}
new field ---------- ]}
]
} |
ProduceResponse Message
For ProduceResponse, similarly will add CurrentLeader & CurrentLeaderBroker to NodeEndpoints to convey new leader info. Similarly these fields are tagged and don’t require a version bump.
...
| Code Block |
|---|
{
"apiKey": 0,
"type": "response",
"name": "ProduceResponse",
"validVersions": "0-9",
"flexibleVersions": "9+",
"fields": [
{ "name": "Responses", "type": "[]TopicProduceResponse", "versions": "0+",
"about": "Each produce response", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName", "mapKey": true,
"about": "The topic name" },
{ "name": "PartitionResponses", "type": "[]PartitionProduceResponse", "versions": "0+",
"about": "Each partition that we produced to within the topic.", "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 error." },
{ "name": "BaseOffset", "type": "int64", "versions": "0+",
"about": "The base offset." },
{ "name": "LogAppendTimeMs", "type": "int64", "versions": "2+", "default": "-1", "ignorable": true,
"about": "The timestamp returned by broker after appending the messages. If CreateTime is used for the topic, the timestamp will be -1. If LogAppendTime is used for the topic, the timestamp will be the broker local time when the messages are appended." },
{ "name": "LogStartOffset", "type": "int64", "versions": "5+", "default": "-1", "ignorable": true,
"about": "The log start offset." },
{ "name": "RecordErrors", "type": "[]BatchIndexAndErrorMessage", "versions": "8+", "ignorable": true,
"about": "The batch indices of records that caused the batch to be dropped", "fields": [
{ "name": "BatchIndex", "type": "int32", "versions": "8+",
"about": "The batch index of the record that cause the batch to be dropped" },
{ "name": "BatchIndexErrorMessage", "type": "string", "default": "null", "versions": "8+", "nullableVersions": "8+",
"about": "The error message of the record that caused the batch to be dropped"}
]},
{ "name": "ErrorMessage", "type": "string", "default": "null", "versions": "8+", "nullableVersions": "8+", "ignorable": true,
"about": "The global error message summarizing the common root cause of the records that caused the batch to be dropped"},
// ---------- Start new field ----------
{ "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
"versions": "9+", "taggedVersions": "9+", "tag": 0, "fields": [
{ "name": "LeaderId", "type": "int32", "versions": "9+", "default": "-1", "entityType": "brokerId",
"about": "The ID of the current leader or -1 if the leader is unknown."},
{ "name": "LeaderEpoch", "type": "int32", "versions": "9+", "default": "-1",
"about": "The latest known leader epoch"} "The latest known leader epoch"}
]}
// ---------- End new field ----------
]},
]},
{ "name": "CurrentLeaderBrokerThrottleTimeMs", "type": "BrokerInfoint32", "versions": "91+", "taggedVersionsignorable": "9+"true, "tagdefault": 1"0",
"about": "CurrentLeader's broker information.",
"fields": [
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." },
// ---------- Start new field ----------
{ "name": "LeaderHostNodeEndpoints", "type": "string[]NodeEndpoint", "versions": "9+", "taggedVersions": "12+", "versionstag": "9+"0,
"about": "TheEndpoints leaderfor hostname." },
all current-leaders enumerated in PartitionData.", "fields": [
{ "name": "LeaderPortNodeId", "type": "int32", "versions": "9+",
"mapKey": true, "entityType": "brokerId", "about": "The leader port ID of the associated node." },
{ "name": "LeaderRackHost", "type": "string", "versions": "9+", "ignorable": true, "default": "null",
"about": "The rack of the leader, or null if it has not been assigned to a racknode's hostname." },
{ ]}
// ---------- End new field ----------
]}"name": "Port", "type": "int32", "versions": "9+",
"about": "The node's port." },
]},
{ "name": "ThrottleTimeMsRack", "type": "int32string", "versions": "19+", "nullableVersions": "9+", "ignorable": true, "default": "0null",
"about": "The rack duration in milliseconds for which the request was throttled dueof the node, or null if it has not been assigned to a quota violation, or zero if the request did not violate any quota." }rack." }
]}
// ---------- End new field ----------
]
} |
Compatibility, Deprecation, and Migration Plan
...