Versions Compared

Old Version 3

changes.mady.by.user Thomas Scott

Saved on

compared with

New Version Current

changes.mady.by.user Thomas Scott

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Table of Contents

Status

Current state: Under DiscussionAccepted

Discussion thread: here 

JIRA: KAFKA-12541 

...

We will add the new offset spec in org.apache.kafka.clients.admin.OffsetSpec:

public class OffsetSpec {

    public static class EarliestSpec extends OffsetSpec { }

    public static class LatestSpec extends OffsetSpec { }

    public static class MaxTimestampSpec extends OffsetSpec { } // this is new

....

    /**
    * Used to retrieve the offset with the largest timestamp of a partition
    * as message timestamps can be specified client side this may not match
    * the log end offset returned by LatestSpec
    */
    public static OffsetSpec maxTimestamp() { // this is new
        return new MaxTimestampSpec();
    }

....

We will need to bump the ListOffsets API version to do this to ensure that requests made to earlier brokers that do not understand this specification are failed. This would be enforced in ListOffsetsRequest.Builder with something similar to the below (where 3 is the new version):

public static Builder forMaxTimestamp(IsolationLevel isolationLevel) {
  return new Builder(3, ApiKeys.LIST_OFFSETS.latestVersion(), CONSUMER_REPLICA_ID, isolationLevel);
}

Proposed Changes

OffsetSpecs are mapped to Long values for timestamp fetches in org.apache.kafka.clients.admin.KafkaAdminClient:

...

The only behaviour changes created by this KIP are for requests to AdminClient.listOffsets that are looking for a timestamp of -3L. As this is not a valid timestamp there should be no migration, compatibility or deprecation concerns. As mentioned in public interfaces,  this functionality is not backwards compatible and attempt to fetch by max timestamp on earlier brokers will be failed.

Rejected Alternatives

  • Create a new API to retrieve the maximum timestamp record information, e.g. AdminClient.getMaxTimestampAndOffset(TopicPartition topicpartition). Rejected as AdminClient.listOffsets is already a good match for this functionality.
  • Extend FetchedTimestampAndOffset to include a timestamp type - timestamp type (log append time or creation time) is a useful piece of information to return in any offset fetching method. It was considered to enrich the listOffsets API with this information as part of this KIP. However, this would involve significant changes to the existing API and for all but some corner cases (where the timestamp type used for the topic changes over time) the timestamp type can be fetched using the AdminClient (describeConfigs) at the topic level.