Confluent-kafka-go: Setting offsets

[gsquire]

I am writing a tool similar to kafkacat due to certain deserialization requirements. I would like to be able to specify an offset relative to the tail of a topic. However when using kafka.OffsetTail, I get an invalid offset when trying to make a new consumer.

I looked at the source for kafkacat and saw that it uses the same function in C as in cgo that I thought would work for Go. I also tried the go-kafkacat example with the -o option and had the same result.

Am I missing something obvious?

[edenhill]

Can you provide an example how you are using OffsetTail?

[gsquire]

Sure, here is what I have:

var offset int64 = 10
delta, _ := kafka.NewOffset(offset)
initialOffset = kafka.OffsetTail(delta).String()

[edenhill]

What are you doing with initialOffset later on, which API are you passing it to?

[gsquire]

I thought I could pass it to a kafka.ConfigMap under the default.topic.config key as such:

"default.topic.config": kafka.ConfigMap{"auto.offset.reset": initialOffset}

[edenhill]

auto.offset.reset will only take a list of logical names: earliest, latest, error, etc.., not an arbitrary (albeit logical) offset.

But if you want to start consuming at a certain position you may set the .Offset field of the TopicPartition (that you pass to Assign())

[gsquire]

I remember trying that as well but eventually see that the topic partitions don't get updated with this snippet:

case kafka.AssignedPartitions:
    newPartitions := make([]kafka.TopicPartition, 0, len(e.Partitions))
    for _, part := range e.Partitions {
    part.Offset.Set(initialOffset)
    newPartitions = append(newPartitions, part)
    }
    fmt.Printf("partitions are at %+v\n", newPartitions)
    consumer.Assign(newPartitions)
    consumer.Commit()

But then the output is this:

partitions are at [redacted[0]@unset redacted[1]@unset]

[edenhill]

Make sure to set the configuration property go.application.rebalance.enable to true

[edenhill]

Here's a complete example based on examples/consumer_example:

https://gist.github.com/edenhill/9dfc019b980a5eb3365c84524a1f12b0

[gsquire]

Thanks so much for the help. The only difference I found with your example and mine is that you do not use the Set function for a kafka.Offset. I tried the direct assignment and it is showing that it gets set now.

Might the Set function have a bug?

[edenhill]

Yeah, Offset.Set() only supports the canonical names for fixed logical offsets:
https://github.com/confluentinc/confluent-kafka-go/blob/master/kafka/kafka.go#L174
It converts those string names to their int counterparts.

So go with a direct Offset = assignment now and you should be fine.

[gsquire]

I see. I thought the call to NewOffset would handle it just fine. Considering it's working now, I will close this. Thanks again!

[JeanMertz]

Sorry for digging up such an old topic, but I was looking for the same capability, and have a follow-up question:

The description for auto.offset.reset states:

Action to take when there is no initial offset in offset store or the desired offset is out of range

My question is, I want to be able for a new consumer group to determine the offset, also by setting the tail offset, _but_ I only want this to happen when there is no offset stored at Kafka already. So: I want it to start at -5, but the next time the consumer connects to Kafka with the same group ID, I want it to continue where it left off, and not start at -5 again.

I looked at your example at https://gist.github.com/edenhill/9dfc019b980a5eb3365c84524a1f12b0#file-tailing_consumer-go-L75-L83, and it is not clear to me that that is what will happen in this case. It seems that _every time_ there is an "assigned partitions" event, you set the offset to -5, but will the Kafka broker simply ignore that value if any other offset is already known at the broker for that group, or will I have to build in some logic to only assign the offset when no offset is known, and if so, how could I detect this?

I did find this in librdkafka:

https://github.com/edenhill/librdkafka/blob/6c722e5d411882283e5687b3e214075804e782ae/src/rdkafka.h#L2394

But it's still unclear to me how to determine when to set the offset manually (the first time basically) and when to not change the offset.

[JeanMertz]

To add to my earlier comment, I did some testing using your example, and indeed, it sets the offset to -5, even if there's already an offset known at the broker (which makes sense, given what the code does).

So then I figured I might want to check for an existing offset, and only apply the tailing offset if none is known yet, but it looks like the offset isn't send as part of the assigned partition event, as described here:

https://github.com/confluentinc/confluent-kafka-go/issues/37#issuecomment-274706136

[edenhill]

You'll want to use Committed(partitions) from your AssignedPartitions handler to get the currently committed offsets on the broker, or OffsetInvalid if there were no previously committed offset for a partition.