...

In this KIP we propose to add version field in the MetadataResponse and UpdateMetadataRequest so that client can refresh metadata if the incoming metadata is older than the existing metadata in its cache.

Public Interfaces

Zookeeper

Add znode /metadata_epoch with the following json format1) Add leader_epoch field to MetadataResponse

MetadataResponse{
  "version" : int32,
  "metadata_epoch" : int64
}

Request/Response protocol

1) Add metadata_epoch field to UpdateMetadataRequest

UpdateMetadataRequest => controller_id controller_epoch metadata_epoch partition_states live_brokers=> throttle_time_ms brokers cluster_id controller_id topic_metadata
  throttle_time_ms => int32
  brokers => [MetadatBroker]
  cluster_id => nullable_str
  controller_id => int32
  topiccontroller_metadataepoch => [TopicMetadata]
 int32
TopicMetadata => topic_error_code topic is_internal partition_metadata
  topic_error_code_epoch => int16
int64  topic => str<-- New
  ispartition_internalstates => boolean[UpdateMetadataRequestPartitionState]
  partitionlive_metadatabrokers => [PartitionMetadata]
 
PartitionMetadata UpdateMetadataRequestBroker]

2) Add metadata_epoch field to MetadataResponse

MetadataResponse => partitionthrottle_errortime_codems brokers partitioncluster_id leader replicas isr offline_replicascontroller_id metadata_epoch topic_metadata
  partitionthrottle_errortime_codems => int16
  partition_id => int32
  leader => int32
  replicasbrokers => [int32MetadatBroker]
  isrcluster_id => [int32]nullable_str
  offlinecontroller_replicasid => [int32]
  leadermetadata_epoch => int32int64   <-- New
  topic_metadata    <-- NEW

=> [TopicMetadata]

3) Add metadata2) Add leader_epoch field to OffsetCommitRequest

OffsetCommitRequest => group_id generation_id memeber_id retention_time topics
  group_id => str
  generation_id => int32
  member_id => str
  retention_time => int64
  topics => [OffsetCommitRequestTopic]
 
OffsetCommitRequestTopic => topic partitions
  topic => str
  partitions => [OffsetCommitRequestPartition]
 
OffsetCommitRequestPartition => partition offset metadata_epoch metadata
  partition => int32
  offset => int64
  leadermetadata_epoch => int32int64      <-- NEW
  metadata => nullable_str

34) Add leadermetadata_epoch field epoch field to OffsetFetchResponse

OffsetFetchResponse => throttle_time_ms response error_code
  throttle_time_ms => int32
  responses => [OffsetFetchResponseTopic]
  error_code => int16
 
OffsetFetchResponseTopic => topic partition_responses
  topic => str
  partition_responses => [OffsetFetchResponsePartition]
 
OffsetFetchResponsePartition => partition offset metadata_epoch metadata error_code
  partition => int32
  offset => int64
  leadermetadata_epoch => int32int64            <-- NEW
  metadata => nullable_str
  error_code => int16
 

Offset topic schema

Add metadata_epoch field 4) Add leader_epoch field to the schema of the offset topic value. 

OFFSET_COMMIT_VALUE_SCHEMA => offset leadermetadata_epoch metadata commit_timestamp expire_timestamp 
  offset => int64
  leadermetadata_epoch => int32int64     <-- NEW
  metadata => str
  commit_timestamp => int64
  expire_timestamp => int64

...

Consumer API

51) Add error INVALID_LEADERMETADATA_EPOCH. This will be is a non-retriable error which and it may be thrown from consumer's API.

2) Add the following two methods to the interface org.apache.kafka.clients.consumer.Consumer

// This method returns the current metadata_epoch in the Consumer. It will be -1 if consumer has never received any MetadataResponse.
public long metadataEpoch()
 
// This method will block until either the timeout has reached or the consumer has received a MetadataResponse whose metadata_epoch >= minMetadataEpoch
public boolean waitForMetadataUpdate(long minMetadataEpoch, long timeout)

Proposed Changes

1) Controller metadata update

Every time controller sends UpdateMetadataRequest, the controller will read the metadata_epoch in the znode /metadata_epoch by 1 and includes the new value in UpdateMetadataRequest

2) Client's Metadata refresh

After client receives MetadataResponse from a broker, it compares with the MetadataResponse with the cached metadata to check whether the MetadataResponse is outdated. The MetadataResponse is outdated if any of the following conditions are true:- Across all those partitions that the client is interested in, there exists a partition whose leader_epoch in the MetadataResposne < MetadataResponse with the leadermetadata_epoch in the cached metadata
- Across all those partitions that the client is interested in, there exists a partition A which is found in the cached metadata but not in the MetadataResponse. And there exists another partition B in MetadataResponse where B.topic == A.topic. This is needed for the partition expansion scenario.

The client will be forced to refresh metadata again with the existing backoff mechanism if the MetadataResponse is determined to be outdated.

Note that producer is interested in all partitions. Consumers can potentially be interested in only partitions that it has explicitly subscribed to. The purpose of checking only a subset of partitions is an optimization which aim to avoid unnecessary metadata refresh when the metadata is only outdated for partitions not needed by client.

last-accepted MetadataResponse. Client will consider the MetadataResponse as outdated and re-fetch Metadata again with the exiting backoff mechanism if the metadata_epoch in the MetadataResponse is smaller than the metadata_epoch in the last-accepted MetadataResponse.

32) Offset commit

When consumer commits offset, it looks up leaderincludes the metadata_epoch of the partition in the cached metadata and includes this value in the OffsetCommitRequest. The leader_epoch will included in the message appended to the offset topic.from the last-accepted MetadataResponse in the OffsetCommitRequest.

When coordinator receives the OffsetCommitRequest, for each partition in the OffsetCommitRequest, it will additionally check whether the leadermetadata_epoch in epoch in the request >= leader epoch in metadata_epoch in the last commit. If not, the offset for this partition is not committed and the error for that partition in the OffsetCommitResponse will be INVALID_LEADERMETADATA_EPOCH. If yes, both the offset and the metadata_epoch for this partition will be written to the offset topic.

34) Offset fetch

After consumer receives OffsetFetchResponse, it remembers the leadermetadata_epoch for each partition it needs to consume. Then the consumer needs to refresh metadata repeatedly until the leadermetadata_epoch in from MetadataResponse >= the cached metadata >= the leader_epoch in OffsetFetchResponse for all partitions it wants to consume. Note that these logic are all hidden from user and the leader_epoch will not be exposed to user via consumer's public API (e.g. OffsetAndMetadata)partitions in the OffsetFetchRespons.

For existing version of the offset topic, leadermetadata_epoch will epoch will not be available in the value of the offset topic message. We will use leadermetadata_epoch epoch = -1 to indicate the missing leadermetadata_epoch. In this case leadermetadata_epoch in epoch in any MetadataResponse will be larger than the leadermetadata_epoch epoch = -1 and the consumer behavior will be the same as it is now.

5) Consumer initialization if offset is stored externally.

After getting the offset to be stored externally, user needs to additionally use the newly-added API metadataEpoch() to get the metadata_epoch in the consumer. This metadata_epoch should be stored externally as well. Before the next consumer seeks to the offset that is stored externally, this consumer should first read the metadata_epoch from the external store and call waitForMetadataUpdate() to make sure that the consumer receives MetadataResponse whose metadata_epoch >= previously-stored metadata_epoch.

Compatibility, Deprecation, and Migration Plan

The KIP changes the inter-broker protocol. Therefore the migration requires two rolling bounce. In the first rolling bounce we will deploy the new code but broker will still communicate using the existing protocol. In the second rolling bounce we will change the config so that broker will start to communicate with each other using the new protocol.

Rejected Alternatives

- Use a global per-metadata version.

 This can be a bit more complicated by introducing a new state in Kafka. leader_epoch is an existing state we already maintain in zookeeper. By using per-partition leader_epoch the client will only be forced to re-fresh metadata if the MetadataResponse contains out-dated metadata for those partitions that the client is interested in.