Status

Current state: Under DiscussionReleased in 4.1

Discussion thread: https://lists.apache.org/thread/gyvfrysoodymgxgwnh4x2g2fs6y5247g

Vote thread: https://lists.apache.org/thread/fb5xxw9313vrljsfqsp2b2cd6gkfldjo

JIRA:

...

KRaft is not able to determine if it should park FETCH requests based on HWM. The KRaft leader doesn't know what HWM was replicated to the remote replicas. Because of this, if the HWM has changed and there is no new data after the FetchOffset, KRaft parks the FETCH request delaying the HWM replication by at most 500ms. 500ms is the maximum amount of time that the KRaft leader will park FETCH requests if there are no records after the FetchOffset.

Replica Manager and ISR Partitions

The broker replica manager has a similar issue with HWM replication. Features like fetch-from-follower need to need to know the latest HWM to allow Kafka consumer to FETCH to the HWM. If there are delays, in the HWM replication those delays will show on applications that care about end-to-end latency between the Kafka producer and Kafka consumer.

Public Interfaces

Public Interfaces

This KIP proposes adding the replica's HWM to the FETCH request This KIP proposes adding the replica's HWM to the FETCH request so that the replica receiving the FETCH request can use the remote replica's HWM to determine if FETCH requests should be completed immediately or parked until there is new data or a new HWM.

Fetch

Metadata Version

A new metadata version will be added. The replica fetcher with send the new FETCH RPC version (18) when the metadata version has been finalized to this new metadata version.

Fetch

Request

Add a version 18 of the RPC. Add the field HighWatermark which the replica sets to its known HWM. Set to -1 if the HWM is not known. The default value for the HWM is the largest int64 or 9223372036854775807.

...

If the response is empty and the remote replica's HWM is greater than or equal to the leader's HWM, the FETCH request is parked.

When the leader's HWM changes, the leader may complete all parked FETCH requests.

Replica manager

The current implementation parks the FETCH request if all of these conditions are true: fetch request has a wait time, fetch request requires data, fetch request doesn't have enough data to respond, no errors happened while reading data, no diverging epoch and no preferred read replica.

.

When the leader's HWM changes, the leader may complete all parked FETCH requestsIn addition to these conditions, the replica manager may also park fetch requests if the remote replica's HWM is greater than or equal to the local replica's HWM.

Sending

When sending FETCH requests the replica will include its HWM in the FETCH request's HighWatermark field. If the remote replica support version 18 of the RPC the HWM will be serialized. If the remote replica doesn't support version 18 the HighWatermark field in the field will ignored and not serialized.

Metadata Version

The inter broker FETCH version is determine using the MV. A new metadata version will be added that will allow brokers to send FETCH version 18 when this new metadata version is the finalized version.

Replica Fetcher and Kafka Consumers

Kafka consumer and the replica fetcher should not set the HighWatermark field. The broker and replica manager will ignore the HighWatermark field if the client is a Kafka Consumer. The replica manager only allows Kafka consumer to fetch offsets that are less than the HWM or stable offset. Kafka broker already has mechanism for completing those requests when the HWM or Stable offset changes.specified.

Compatibility, Deprecation, and Migration Plan

...

For KRaft, this feature will mainly be tested using the KRaft protocol unittests as supported in KafkaRaftClientFetchTest.For Replica manager, this feature will be tested using unittests in ReplicaManagerTest.