Status

Current state: Under Discussion

Discussion thread: here

JIRA:

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

Motivation

It is common practice to monitor, and potentially programmatically react to, consumer lag. The lag is calculated as the difference between the HighwaterMarkOffset and the ConsumerOffset. Currently the only way to get the HighwaterMarkOffset for a partition is via a FetchRequest/Response. However, in this case we do not actually want to fetch any records. Adding the HighwaterMarkOffset to the OffsetFetchResponse allows for simple calculation and reaction to consumer lag.

Public Interfaces

The proposal is to add HighwaterMarkOffset to the OffsetFetchResponse:

OffsetFetchResponse => [TopicName [Partition Offset Metadata ErrorCode HighwaterMarkOffset]]
  TopicName => string
  Partition => int32
  Offset => int64
  Metadata => string
  ErrorCode => int16
  HighwaterMarkOffset => int64

Proposed Changes

The full description is well covered in the Motivation and Public Interfaces section.

We plan to bump the version of OffsetFetchResponse and OffsetFetchRequest - OffsetFetchRequest does not actually change, but current Api design deduces whether clients support the newer response versions based on the request version.

Since this is an extra field in the response, it will be ignored by older clients and therefore pose no special upgrade challenges.

The implementation will be linked here when a first pass is completed.

Compatibility, Deprecation, and Migration Plan

This change is additive and should have minimal impact. A new version of the OffsetFetchRequest/Response would need to be added. The versioning procedure is defined here:

https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-VersioningandCompatibility

Rejected Alternatives

Currently the existing alternative is to make 2 calls. A FetchRequest and an OffsetFetchRequest.