Kafkajs: I would like to read from offset 0

[esuarezz]

Hi,

when client or backend disconnect I want to consume from beginning the whole information saved and not continue from where stopped,

I thought it would be with "fromBeginning" but not..

[Nevon]

fromBeginning is used when there's no resolved offset to resume from. If you just restart your consumers, they will resume where they left off.

If you want to reset the offsets to the earliest offset, you'll need to use the admin API: https://github.com/tulios/kafkajs#-reset-consumer-group-offsets This could get tricky though, because you would need to do this when your first consumer is created, but any other consumer that joins the group later should not reset the offsets.

What is the use-case you have where you need to do this?

[esuarezz]

It is because when I reset the client, needs to get the info from beginning on kafka again, so I need the consumer(in a BE app) consuming again from scratch and sending to the client.

in this case we have just one consumer and just one application consuming from this

[Nevon]

An easier solution for you then might be to seek to the beginning. For example:

const EARLIEST_OFFSET = -2 // this is a special value

const partitions = [0,1,2]

partitions.forEach((partition) => {
  consumer.seek({ topic: 'example', offset: EARLIEST_OFFSET, partition })
})

This seems like maybe it could be a recurring pattern, so maybe it makes sense to make it simpler.

[esuarezz]

Got this:

(node:4368) UnhandledPromiseRejectionWarning: KafkaJSNonRetriableError: Consumer group was not initialized, consumer#run must be called first

I am init consumer in the beginning.. and it is working fine the consumer -> consuming, but for this functionality getting this exception

[tulios]

@esuarezz just run seek after run, example:

consumer.run({ eachMessage: async ({ topic, message }) => true })
consumer.seek({ topic: 'example', partition: 0, offset: 12384 })

[tulios]

@esuarezz feel free to re-open the issue if the problem persists.
Thanks.