...

OFFSET_COMMIT_VALUE_SCHEMA => offset leader_epoch metadata commit_timestamp expire_timestamp 
  offset => int64
  leader_epoch => int32     <-- NEW
  metadata => str
  commit_timestamp => int64
  expire_timestamp => int64

5) Add error INVALID_LEADER_EPOCH. This will be a non-retriable error which may be thrown from consumer's API.

Proposed Changes

1) Metadata refresh

...

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.

2) Offset commit and fetch

When consumer commits offset, it looks up leader_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.

When coordinator receives the OffsetCommitRequest, for each partition in the OffsetCommitRequest, it will additionally check whether the leader_epoch in the request >= leader epoch in the last commit. If not, the error for that OffsetCommitResponse will be INVALID_LEADER_EPOCH.

3) Offset fetch

After consumer receives OffsetFetchResponse, it remembers the leader_epoch for each partition it needs to consume. Then the consumer needs to refresh metadata until the leader_epoch in 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).

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

...