Versions Compared

Old Version 1

changes.mady.by.user Mickael Maison

Saved on

compared with

New Version 2

changes.mady.by.user Mickael Maison

Saved on

Key

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

Table of Contents

Status

Current state: Draft

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

...

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

Motivation

Resetting the offsets of a consumer group is a relatively common operation. It's usually done to allow an application to reprocess data or otherwise to skip/drop records. Currently (thanks to KIP-122) it can be done using the kafka-consumer-group.sh tool however it would be nice to be able to to it directly with the AdminClient. Applications can more easily depend on the AdminClient API rather than running the ConsumerGroupsCommand class or worse reimplementing the logic.

Additionally, at the moment, the kafka-consumer-group.sh can only run against a single group at a time and needs to starts a new consumer every time which is rather inefficient when resetting many groups. This tool could even be updated afterwards to use the AdminClient instead of having its own logic.

Public Interfaces

We will add 4 new methods to the AdminClient public API:

...

  • CommitOffsetsOptions


    Code Block
    languagejava
    titleCommitOffsetsOptions.java
    /**
 * Options for the {@link AdminClient#commitOffsets(String, org.apache.kafka.clients.consumer.OffsetAndMetadata, CommitOffsetsOptions) call.
 *
 * The API of this class is evolving, see {@link AdminClient} for details.
 */
@InterfaceStability.Evolving
public class CommitOffsetsOptions extends AbstractOptions<CommitOffsetsOptions> {
}


  • ListOffsetsResult


    Code Block
    languagejava
    titleListOffsetsResult.java
    /**
 * The result of the {@link AdminClient#listOffsets(String)} call.
 *
 * The API of this class is evolving, see {@link AdminClient} for details.
 */
@InterfaceStability.Evolving
public class ListOffsetsResult {

    public ListOffsetsResult(Map<TopicPartition, KafkaFuture<ListOffsetsResultInfo>> futures) {}

    /**
     * Return a map from TopicPartition to futures which can be used to retrieve the offsets
     */
    public Map<TopicPartition, KafkaFuture<ListOffsetsResultInfo>> values() {}

    /**
     * Return a future which succeeds only if offsets for all specified partitions have been successfully
     * retrieved.
     */
    public KafkaFuture<Map<TopicPartition, ListOffsetsResultInfo>> all() {}

    static public class ListOffsetsResultInfo {

        ListOffsetsResultInfo(long offset, long timestamp) {}

        long getOffset() {}
        long getTimestamp() {}
    }
}


  • ListOffsetsOptions



    Code Block
    languagejava
    titleListOffsetsOptions.java
    /**
 * Options for {@link AdminClient#listOffsets()}.
 *
 * The API of this class is evolving, see {@link AdminClient} for details.
 */
@InterfaceStability.Evolving
public class ListOffsetsOptions extends AbstractOptions<ListOffsetsOptions> {

    private final long timestamp;
    private final String isolationLevel;

    public static ListOffsetsOptions latestUncommitted() {}

    public static ListOffsetsOptions latestCommitted() {}

    public static ListOffsetsOptions earliestUncommitted() {}

    public static ListOffsetsOptions earliestCommitted() {}

    public ListOffsetsOptions(long timestamp, String isolationLevel) {}

    public long timestamp() {}

    public String isolationLevel() {}
}



Proposed Changes

I propose adding methods to commit offsets for a group in the AdminClient. Alongside, I will add methods to retrieve log offsets (ListOffset API) so users can easily retrieve start and end offsets or offsets for a timestamp for partitions.

...

  • public CommitOffsetsResult commitOffsets(String groupId, Map<TopicPartition, OffsetAndMetadata> offsets) {}
  • public abstract CommitOffsetsResult commitOffsets(String groupId, Map<TopicPartition, OffsetAndMetadata> offsets, CommitOffsetsOptions options);
  • public ListOffsetsResult listOffsets(Collection<TopicPartition> topicPartitions) {}
  • public abstract ListOffsetsResult listOffsets(Collection<TopicPartition> topicPartitions, ListOffsetsOptions options);

Compatibility, Deprecation, and Migration Plan

No impact or migration plan required as this proposal is only adding new methods.

Rejected Alternatives

None