...

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

3) Add the following method and class methods to the interface org.apache.kafka.clients.consumer.Consumer

void seek(TopicPartition partition, long offset, String internalMetadata);

 
OffsetAndMetadata positionWithInternalMetadata(TopicPartition partition);

4) Add two fields to field internalMetadata to the class org.apache.kafka.clients.consumer.OffsetAndMetadata

public class OffsetAndMetadata {
 
  // This is the newly-added internal metadata used by Kafka client implementation. User should not need to interpret this value.
  private final String internalMetadata;
 
  // The existing metadata variable used by user. Kafka client or broker should not need to interpret this value.
  private final String metadata;
 
  .... // Other existing variables and methods
 
}

This field should be a json formatted string that currently encodes leader_epoch and topic_epoch with the following format:

{
  "version" : 1,
  "topic_epoch" : int32,
  "leader_epoch" : int32
}

We can evolve the format of this field in the future as needed. This should only be used by Kafka client implementation and user should not need to interpret this string.

Proposed Changes

1) Topic creation and partition expansion.

...