Kafka reset consumer offset java

Kafka new consumer: (re)set and commit offsets with using assign, not subscribe

Using the new Kafka Java consumer api, I run a single consumer to consume messages. When all available messages are consumed, I kill it with kill -15 . Now I would like to reset the offsets to start. I would like to avoid to just use a different consumer group. What I tried is the following sequence of calls, using the same group as the consumer that just had finished reading the data.

assign(topicPartition); OffsetAndMetadata om = new OffsetAndMetadata(0); commitSync(Collections.singletonMap(topicPartition, 0)); 

ERROR internals.ConsumerCoordinator: Error UNKNOWN_MEMBER_ID occurred while committing offsets for group queue Exception in thread "main" org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed due to group rebalance at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:552) 

Is it in principle wrong to combine assign with commitSync , possibly because only subscribe and commitSync go together? The docs only say that assign does not go along with subscribe , but I thought this applies only in one consumer process. (In fact I was even hoping to run the offset-reset consumer while the other consumer is up, hoping that the other one might notice the offset change and start over again. But shutting it down first is fine too.) Any ideas?

Источник

Is there a way to reset offsets of a Kafka consumer group through an API?

I have a use case where there is a consumer group that is consuming messages. I want to build an API to modify its offset. So, when the endpoint is called with the offset I have to change the offset of the consumer group. I am using SpringBoot and the consumer is built using Spring Kafka. Thanks in advance.

3 Answers 3

This is a solution via CLI:

List the topics to which the group is subscribed:

kafka-consumer-groups --bootstrap-server --group --describe 

Note the values under «CURRENT-OFFSET» and «LOG-END-OFFSET». «CURRENT-OFFSET» is the offset where this consumer group is currently at in each of the partitions.

Reset the consumer offset for a topic (preview):

kafka-consumer-groups --bootstrap-server --group --topic --reset-offsets --to-earliest 

This will print the expected result of the reset, but not actually run it.

Reset the consumer offset for a topic (execute):

kafka-consumer-groups --bootstrap-server --group --topic --reset-offsets --to-earliest --execute 

This will execute the reset and reset the consumer group offset for the specified topic back to 0.

Repeat 1 to check if the reset is successful

Spring for Apache Kafka provides some convenience mechanisms for performing seeks, either during application initialization, or at any time thereafter.

The simplest is to have your listener extend AbstractConsumerSeekAware or implement ConsumerSeekAware .

How should I seek based on API? What method should I call to seek to a particular offset? Can you give a code sample, it would be of great help. Thanks

See the documentation I pointed you to and come back if you have a more specific question; there are several examples in answers on this site; e.g. stackoverflow.com/questions/66163492/… stackoverflow.com/questions/59682801/… search for AbstractConsumerSeekAware .

To reset an offset, the code has to get the consumer by group id, get started, and stopped. A client has to be created (while consumer is stopped). Once the client has been created, the code has to connect, reset the offset, and then disconnect (as posted below). Once it has reset the consumer group has to be restarted again.

The code below below resets the offset. The key is to get the consumer instance to be able to stop and start the consumers because it cannot reset if a consumer is running. We have this running from the command line which is nice. Hopefully this helps.

return kafka.getConsumer( args.groupId ) .then(consumerInstance => < consumer = consumerInstance; return consumer.run(); >).then(()=> < return consumer.stop(); >).then(() => < // reset kafka client offset >)).then(()=> < return consumer.run(); >). // reset kafka client offset const < groupId, topic >= args; const admin = createKafkaClient().admin(); return admin.connect() .then(() => admin.resetOffsets(< groupId, topic, earliest: true, >) .then(() => admin.disconnect(); 

Источник

How does spring.kafka.consumer.auto-offset-reset works in spring-kafka

At this case @KafkaListener method is invoked for all entries. But expected result was that @KafkaListener method is invoked only for latest 3 options I send. I tried to use another option:

spring.kafka.consumer.auto-offset-reset=earlisest 

P.S.

@SpringBootApplication public class Application implements CommandLineRunner < public static Logger logger = LoggerFactory.getLogger(Application.class); public static void main(String[] args) < SpringApplication.run(Application.class, args).close(); >@Autowired private KafkaTemplate template; private final CountDownLatch latch = new CountDownLatch(3); @Override public void run(String. args) throws Exception < this.template.send("spring_kafka_topic", "foo1"); this.template.send("spring_kafka_topic", "foo2"); this.template.send("spring_kafka_topic", "foo3"); latch.await(60, TimeUnit.SECONDS); logger.info("All received"); >@KafkaListener(topics = "spring_kafka_topic") public void listen(ConsumerRecord cr) throws Exception < logger.info(cr.toString()); latch.countDown(); >> 

Update:

Behaviour doesn’t depends on
spring.kafka.consumer.auto-offset-reset it is only depends on spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit if I set spring.kafka.consumer.enable-auto-commit=false — I see all records. if I set spring.kafka.consumer.enable-auto-commit=true — I see only 3 last records. Please clarify menaning of spring.kafka.consumer.auto-offset-reset property

Источник

How to reset Kafka offsets to match tail position?

We’re using Storm with Kafka and ZooKeeper. We had a situation where we had to delete some topics and recreate them with different names. Our Kafka spouts stayed the same, aside from now reading from the new topic names. However now the spouts are using the offsets from the old topic partitions when trying to read from the new topics. So the tail position of my-topic-name partition 0 will be 500 but the offset will be something like 10000. Is there a way to reset the offset position so it matches the tail of the topic?

Your kafka offsets are stored in a znode in Zookeeper. So you can reset them via zookeeper cli, they should be under /consumers. You can read them using the kafka metadata API. I have been working on a project where we actually override the built in offsets mechanism and store the offsets in HBase, and use multiversioning to story historical offsets. We basically have a transaction history of offsets that we can roll back by deleting versions of the offset for each consumer group/topic/partition combination.

@dsch that only true for pre 0.9 version of Kafka. Since 0.9 offsets are stored in a Kafka topic __consumer_offsets

1 Answer 1

There a multiple options (as Storm’s KafkaSpout does not provide any API to define the starting offset).

1. If you want to consumer from the tail of the log you should delete old offsets
   • depending on you Kafka version
     • (pre 0.9) you can manipulate ZK (which is a little tricky)
     • (0.9+) or you try do delete the offset from the topic __consumer_offsets (which is also tricky and might delete other offset you want to preserve, too)
   • if no offsets are there, you can restart your spout with auto offset reset policy «latest» or «largest» (depending on you Kafka version)
2. as an alternative (which I would recommend), you can write a small client application that uses seek() to manipulate the offset in the way you need them and commit() the offsets. This client must use the same group ID as you KafkaSpout and must subscribe to the same topic(s). Furthermore, you need to make sure that this client application is running a single consumer group member so it get’s all partitions assigned.
   • for this, you an either seek to the end of the log and commit
   • or you commit an invalid offset (like -1) and rely on auto offset reset configuration»latest» or «largest» (depending on you Kafka version)

For Kafka Streams, there is a «Application Reset Tool» that does a similar thing to manipulate committed offsets. If you want to get some details, you can read this blog post http://www.confluent.io/blog/data-reprocessing-with-kafka-streams-resetting-a-streams-application/

(disclaimer: I am the author of the post and it is about Kafka Streams — nevertheless, the underlying offset manipulation ideas are the same)

Linked

Related

Hot Network Questions

Subscribe to RSS

To subscribe to this RSS feed, copy and paste this URL into your RSS reader.

Site design / logo © 2023 Stack Exchange Inc; user contributions licensed under CC BY-SA . rev 2023.7.17.43537

By clicking “Accept all cookies”, you agree Stack Exchange can store cookies on your device and disclose information in accordance with our Cookie Policy.

Источник

How to change or reset consumer offset in Kafka?

This is a very common use case. Consumer offset is used to track the messages that are consumed by consumers in a consumer group. A topic can be consumed by many consumer groups and each consumer group will have many consumers. A topic is divided into multiple partitions.

A consumer in a consumer group is assigned to a partition. Only one consumer is assigned to a partition. A consumer can be assigned to consume multiple partitions.

Consumer offset is managed at the partition level per consumer group.

Check our post titled “What is consumer offset and the purpose of consumer offset in Kafka?” to learn more about consumer offset

Why change consumer offset?

In some scenarios, consumers which consumed the messages from a Kafka partition could have resulted in errors and the consumption would have been incomplete. In such cases of consumption failures you may have a need to re-consume the messages which were previously consumed. In such instances you would have to reset the consumer offset to an earlier offset.

How to find the current consumer offset?

Use the kafka-consumer-groups along with the consumer group id followed by a describe.

kafka-consumer-groups.sh --bootstrap-server --group --describe

You will see 2 entries related to offsets – CURRENT-OFFSET and LOG-END-OFFSET for the partitions in the topic for that consumer group. CURRENT-OFFSET is the current offset for the partition in the consumer group.

How to change consumer offset?

Use the kafka-consumer-groups.sh to change or reset the offset. You would have to specify the topic, consumer group and use the –reset-offsets flag to change the offset.

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliest --all-topics --execute

Consumer offset reset options

–reset-offsets has several options and you can pick the correct one based on your needs.

–shift-by

Use shift-by to move the offset ahead or behind. It can take both +ve or -ve number.

Reset the offset by incrementing the current offset position by 10

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --shift-by 10 --topic sales_topic --execute

Reset the offset by decrementing the current offset position by 10

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --shift-by -10 --topic sales_topic --execute

–to-datetime

Reset offsets to offset from datetime. Format: ‘YYYY-MM-DDTHH:mm:SS.sss’

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-datetime 2020-11-01T00:00:00Z --topic sales_topic --execute

–to-earliest

Reset offsets to earliest (oldest) offset available in the topic.

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-earliest --topic sales_topic --execute

–to-latest

Reset offsets to latest (recent) offset available in the topic.

kafka-consumer-groups.sh --bootstrap-server kafka-host:9092 --group my-group --reset-offsets --to-latest --topic sales_topic --execute

Источник