...

JIRA:

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

Motivation

Cluster Metadata and KRaft

Many users have observed that their admin operations can take up to 500 milliseconds, after they complete, to show up in other clients. See

...

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.

This KIP proposes adding the replica's HWM to the FETCH request so that the KRaft leader 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.

Public Interfaces

Fetch

Request

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

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

Request

Add a version 18 of the RPC. Add 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.

...

In 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.

Kafka Consumers

Kafka consumer 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.

Compatibility, Deprecation, and Migration Plan

...