THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
In order to support fetching from the closest replica, we need 1) to add support in the Fetch protocol for follower fetching, and 2) some way for the consumer to determine its distance to the replicas of a partitiona mechanism to find the "closest" replica to a given consumer.
Follower Fetching
We propose to extend the fetch protocol to allow consumer fetching from any replica. Just as when fetching from the leader, the replication protocol will guarantee that only committed data is returned to the consumer. This relies on propagation of the high watermark through the leader's fetch response. Due to this propagation delay, follower fetching may have higher latency. Additionally, follower fetching introduces the possibility of spurious out of range errors if a fetch is received by an out-of-sync replica. The diagram below illustrates the different cases that are possible.
...
Case 2 (unavailable offset): The offset is not available locally, but is known to be committed. This can occur if a fetch is received by an out-of-sync replica. Even out-of-sync replicas will continue fetching from the leader and they may have received a high watermark value which is larger than its log end offset. So a fetch at an offset between the log end offset and the high watermark received from the leader is known to be committed though the replica does not have the data.
In generalIdeally, consumers should fetch only from the ISR, but it is not possible to prevent fetching from outside the ISR because it takes time for the ISR state to propagate to each replica. We propose here to also return OFFSET_NOT_AVAILABLE for this case, which will cause the consumer to retry. In the common case, we expect the out-of-sync replica to return to the ISR after a short time. On the other hand, if the replica remains out-of-sync for a longer duration, the consumer will have a chance to find a new replica after its metadata refreshsome time. We discuss this point in more detail below in the section on how the consumer finds its preferred replica.
...
In fact, the logic to handle this case is already outlined in KIP-320. The consumer will reconcile its current offset with the leader and then will resume fetching. The only change we make here is to request a metadata update if we find that the current offset is still valid on the leader. This gives us a chance to detect any ISR changes before we resume fetching from the same replica.
...
The benefit of letting the broker decide which replica is preferred for a client is that it can take load into account. For example, the broker can round robin between nearby replicas in order to spread the load. There are many such considerations that are possible, so we we propose to allow users to provide a ReplicaSelector plugin to the broker in order to handle the logic. This will be exposed to the broker with the replica.selector.class configuration. In order to make use of this plugin, we will extend the Metadata Fetch API so that clients can provide their own location information. In the response, the broker will indicate a preferred replica to fetch from.
...
Consumer API
We use of the "broker.rack.id" configuration to identify the location of the broker, so we will add the same a similar configuration to the consumer. This will be used in Metadata Fetch requests as documented below.
...
| Configuration Name | Valid Values | Default Value |
|---|---|---|
| client.rack.id | <nullable string> | null |
Broker API
...
Protocol Changes
We will extend the Metadata Fetch API in order to enable selection of the preferred replica. We use of the "client.rack.id" configuration in the consumer that was mentioned above to identify the location of the broker, so we will add a similar configuration to the consumer.
| Code Block |
|---|
MetadataRequest => [TopicName] AllowAutoTopicCreation RackId
TopicName => STRING
AllowAutoTopicCreation => BOOLEAN
RackId => STRING // This is new |
The metadata response will indicate the preferred replica.
{
"validVersions": "0-11",
"fields": [
{
"name": "ReplicaId",
"type": "int32",
"versions": "0+",
"about": "The broker ID of the follower, of -1 if this request is from a consumer."
},
{
"name": "MaxWait",
"type": "int32",
"versions": "0+",
"about": "The maximum time in milliseconds to wait for the response."
},
{
"name": "MinBytes",
"type": "int32",
"versions": "0+",
"about": "The minimum bytes to accumulate in the response."
},
{
"name": "MaxBytes",
"type": "int32",
"versions": "3+",
"default": "0x7fffffff",
"ignorable": true,
"about": "The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored."
},
{
"name": "IsolationLevel",
"type": "int8",
"versions": "4+",
"default": "0",
"ignorable": false,
"about": "This setting controls the visibility of transactional records. Using READ_UNCOMMITTED (isolation_level = 0) makes all records visible. With READ_COMMITTED (isolation_level = 1), non-transactional and COMMITTED transactional records are visible. To be more concrete, READ_COMMITTED returns all data from offsets smaller than the current LSO (last stable offset), and enables the inclusion of the list of aborted transactions in the result, which allows consumers to discard ABORTED transactional records"
},
{
"name": "SessionId",
"type": "int32",
"versions": "7+",
"default": "0",
"ignorable": false,
"about": "The fetch session ID."
},
{
"name": "Epoch",
"type": "int32",
"versions": "7+",
"default": "-1",
"ignorable": false,
"about": "The fetch session ID."
},
{
"name": "Topics",
"type": "[]FetchableTopic",
"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": "FetchPartitions",
"type": "[]FetchPartition",
"versions": "0+",
"about": "The partitions to fetch.",
"fields": [
{
"name": "PartitionIndex",
"type": "int32",
"versions": "0+",
"about": "The partition index."
},
{
"name": "CurrentLeaderEpoch",
"type": "int32",
"versions": "9+",
"default": "-1",
"ignorable": true,
"about": "The current leader epoch of the partition."
},
{
"name": "FetchOffset",
"type": "int64",
"versions": "0+",
"about": "The message offset."
},
{
"name": "LogStartOffset",
"type": "int64",
"versions": "5+",
"default": "-1",
"ignorable": false,
"about": "The earliest available offset of the follower replica. The field is only used when the request is sent by the follower."
},
{
"name": "MaxBytes",
"type": "int32",
"versions": "0+",
"about": "The maximum bytes to fetch from this partition. See KIP-74 for cases where this limit may not be honored."
}
]
}
]
},
{
"name": "Forgotten",
"type": "[]ForgottenTopic",
"versions": "7+",
"ignorable": false,
"about": "In an incremental fetch request, the partitions to remove.",
"fields": [
{
"name": "Name",
"type": "string",
"versions": "7+",
"entityType": "topicName",
"about": "The partition name."
},
{
"name": "ForgottenPartitionIndexes",
"type": "[]int32",
"versions": "7+",
"about": "The partitions indexes to forget."
}
]
},
{
"name": "RackId",
"type": "string",
"versions": "11+",
"default": "",
"ignorable": true,
"about": "Rack ID of the consumer making this request"
}
]
}
|
The fetch response will indicate the preferred replica.
| Code Block |
|---|
{
"apiKey": 1,
"type": "response",
"name": "FetchResponse",
"validVersions": "0-11",
"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": false,
"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": "Topics",
"type": "[]FetchableTopicResponse",
"versions": "0+",
"about": "The response topics.",
"fields": [
{
"name": "Name",
"type": "string",
"versions": "0+",
"entityType": "topicName",
"about": "The topic name."
},
{
"name": "Partitions",
"type": "[]FetchablePartitionResponse",
"versions": "0+",
"about": "The topic partitions.",
"fields": [
{
"name": "PartitionIndex",
"type": "int32",
"versions": "0+",
"about": "The partiiton 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": "Aborted",
"type": "[]AbortedTransaction",
"versions": "4+",
"nullableVersions": "4+",
"ignorable": false,
"about": "The aborted transactions.",
"fields": [
{
"name": "ProducerId",
"type": "int64",
"versions": "4+",
"entityType": "producerId",
"about": "The producer id associated with the aborted transaction."
},
{
"name": "FirstOffset",
"type": "int64",
"versions": "4+",
"about": "The first offset in the aborted transaction."
}
]
},
{
"name": "PreferredReadReplica",
"type": "int32",
"versions": "11+",
"ignorable": true,
"about": "The preferred read replica for the consumer to use on its next fetch request"
},
{
"name": "Records",
"type": "bytes",
"versions": "0+",
"nullableVersions": "0+",
"about": "The record data."
}
]
}
]
}
]
}
|
| Code Block |
MetadataResponse => ThrottleTimeMs Brokers ClusterId ControllerId [TopicMetadata]
ThrottleTimeMs => INT32
Brokers => [MetadataBroker]
ClusterId => NULLABLE_STRING
ControllerId => INT32
TopicMetadata => ErrorCode TopicName IsInternal [PartitionMetadata]
ErrorCode => INT16
TopicName => STRING
IsInternal => BOOLEAN
PartitionMetadata => ErrorCode PartitionId Leader LeaderEpoch Replicas ISR OfflineReplicas
ErrorCode => INT16
PartitionId => INT32
Leader => INT32
LeaderEpoch => INT32
PreferredReadReplica => INT32 // This is new
Replicas => [INT32]
ISR => [INT32]
OfflineReplicas => [INT32] |
The FetchRequest schema has field for the replica id. Consumers typically use the sentinel -1, which indicates that fetching is only allowed from the leader. A lesser known sentinel is -2, which was originally intended to be used for debugging and allows fetching from followers. We propose to let the consumer use this to indicate the intent to allow fetching from a follower. Similarly, when we need to send a ListOffsets request to a follower in order to find the log start offset, we will use the same sentinel for the replica id field.
...