[Question] Start consuming from the end? about franz-go HOT 10 CLOSED

Yep, exactly.

from franz-go.

Are you using a consumer group?

from franz-go.

To start consuming from the end in general, you need the ConsumeResetOffset option:

func ConsumeResetOffset(offset Offset) ConsumerOpt
    ConsumeResetOffset sets the offset to restart consuming from when a
    partition has no commits (for groups) or when beginning to consume a
    partition (for direct partition consuming), or when a fetch sees an
    OffsetOutOfRange error, overriding the default ConsumeStartOffset.

with NewOffset().AtEnd().

This will always start consuming from the end for direct partition consuming, or at the end when a partition is seen for the first time when group consuming.

from franz-go.

from franz-go.

If you're using a consumer group, can you use LogLevelInfo? When you shutdown, is there a log about committing?

What's your NewClient look like, and are you waiting for Close to finish when shutting down?

from franz-go.

Thank you so much for your speedy reply and this is exactly what I needed

from franz-go.

So this is similar to what i am looking for, but I only want to start consuming from the end if the consumer group already doesn't have a current offset.

Essentially trying to change this line:
https://github.com/twmb/franz-go/blob/v1.6.0/pkg/kgo/config.go#L488

to be NewOffset().AtEnd() since once the consumer group exists I don't want to repeatedly start at the end, just the very first time it is initialized and doesn't have any offsets stored.

Is there a way I can pass an offset to ConsumeResetOffset() that will use current offsets if they exist otherwise use the latest offset possible?

from franz-go.

@ekeric13 that's actually how the option works currently -- AtEnd will be used for new groups with no commits. Existing offsets will be used if they are committed. If a client ever receives offset out of range, it will reset to the nearest offset (if you're on a modern broker, it will reset to an exact offset).

from franz-go.

So when creating my client I just need to do:

clientOpts := []kgo.Opt{
        ....
	kgo.ConsumeResetOffset(kgo.NewOffset().AtEnd())
	}
cl, err := kgo.NewClient(clientOpts...)

and it will correctly just use existing offsets if commited?

Likewise I could do:

clientOpts := []kgo.Opt{
        ....
	kgo.ConsumeResetOffset(kgo.NewOffset().AtStart())
	}
cl, err := kgo.NewClient(clientOpts...)

and it will replicate the default behavior: start at the beginning if there are no offsets, otherwise use current committed offsets?

from franz-go.

one thing to keep in mind is that when you use kgo.NewOffset().AtEnd() to connect for the first time, any previous messages that has not been consumed will not show up. Any new messages after you connect will, even if you exit and restart it will.

from franz-go.