Restart processing since last committed offset in Kafka

user_2953724 0
Viewed 1100

Pretext

I have:

  • A connected Kafka consumer
  • The consumer is part of a consumer group.
  • A rebalance listener attached to the consumer
  • Auto-commit is disabled

Additionally, I have a method that takes in two parameters, the consumer, and a rebalance-listener that tracks what partitions have been assigned to the consumer

void aggregateProcessing(ConsumerRecords<String, SomeClass> consumer, RebalanceListener listener)

public class RebalanceListener implements ConsumerRebalanceListener {
    private Set<TopicPartition> assignedPartitions = new LinkedHashSet<>();

    @Override
    public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
       // keep track of assigned partitions
    }   
}

Goal

This method is run on a timer, and its goal is to process records until there are none left to read or up until some max time in all of the partitions.

Since rebalancing could happen in the middle of consumption (after consumer.poll() has fired several times already), I would like to detect this, reset, and restart processing from the last committed offset for all assigned partitions (even if that had already been assigned).

Is there a way to reset the consumer's internal offset for each partition back to the latest committed offset for a list of assigned partitions?

Sidenote

I understand reprocessing for all partitions (instead of just for the ones that were changed) is less efficient than selectively expunging some of the processing, but will probably be significantly easier than tracking what data needs to be expunged when a partition is removed.

Thanks!

apache-kafka java
2 Answers

Yes You Can

You have to track your committed offset manually for this purpose.

  1. Whenever the partitions are reinvoked from the consumer, you have to save the partitions and their committed offsets in DB.

  2. When reassigned to the partitions to the consumer, you have to seek from a specific offset that is stored in your datastore.

Your rebalance listener will listen when these revoke and assign events happens. A sample implementation of the rebalance listener

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
           commitDBTransaction();
    }
    public void onPartitionsAssigned(Collection<TopicPartition>  partitions){
            for(TopicPartition partition: partitions) {
                consumer.seek(partition, getOffsetFromDB(partition));
            }
    }
}
user_128944680

As Kafka documentation described in KafkaConsumer

Offsets and Consumer Position

Kafka maintains a numerical offset for each record in a partition. This offset acts as a unique identifier of a record within that partition, and also denotes the position of the consumer in the partition. For example, a consumer which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5. There are actually two notions of position relevant to the user of the consumer: The position of the consumer gives the offset of the next record that will be given out. It will be one larger than the highest offset the consumer has seen in that partition. It automatically advances every time the consumer receives messages in a call to poll(Duration).

The committed position is the last offset that has been stored securely. Should the process fail and restart, this is the offset that the consumer will recover to. The consumer can either automatically commit offsets periodically; or it can choose to control this committed position manually by calling one of the commit APIs (e.g. commitSync and commitAsync).

So, When you need to start with latest committed offset and you have disabled enable.auto.commit, Then you can manually commit your proceesed message's offset.

can choose to control this committed position manually by calling one of the commit APIs (e.g. commitSync and commitAsync).

Then after restarting and rebalancing Kafka, Consumers will start to consume from last committed (processed) offset.

The above scenario is when you use the Kafka storage for consumer offsets. If you already have offsets you want to start consuming, Then you can control consumer's starting offset by Consumer.Seek() before start consuming.

Controlling The Consumer's Position

In most use cases the consumer will simply consume records from beginning to end, periodically committing its position (either automatically or manually). However Kafka allows the consumer to manually control its position, moving forward or backwards in a partition at will. This means a consumer can re-consume older records, or skip to the most recent records without actually consuming the intermediate records. There are several instances where manually controlling the consumer's position can be useful.

One case is for time-sensitive record processing it may make sense for a consumer that falls far enough behind to not attempt to catch up processing all records, but rather just skip to the most recent records.

Another use case is for a system that maintains local state as described in the previous section. In such a system the consumer will want to initialize its position on start-up to whatever is contained in the local store. Likewise if the local state is destroyed (say because the disk is lost) the state may be recreated on a new machine by re-consuming all the data and recreating the state (assuming that Kafka is retaining sufficient history).

Kafka allows specifying the position using seek(TopicPartition, long) to specify the new position. Special methods for seeking to the earliest and latest offset the server maintains are also available ( seekToBeginning(Collection) and seekToEnd(Collection) respectively).

user_108667980
Related