Committed offset is reset periodically

[n5i]

Description

We have a consumer which reads the last uncommitted message, processes and commits it. It was working fine until we discovered that all messages are periodically re-consumed. It looks like the last offset is reset and consumer.Committed() returns the -1001. It happened every other day without a strong pattern. About 5 days ago the problem was gone.

The consumer code is below.

`// Set up Kafka consumer
consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
"bootstrap.servers": config.KafkaHost,
"group.id": "XXXSync",
"session.timeout.ms": 6000,
"go.events.channel.enable": true,
"enable.auto.commit": false,
"default.topic.config": kafka.ConfigMap{"auto.offset.reset": "earliest"}})
if err != nil {
return fmt.Errorf("xxx")
}
defer consumer.Close()

// Read only from the first partition
var partition int32 = 0

// Get offset of the last committed message
lastOffsets, err := consumer.Committed([]kafka.TopicPartition{{Topic: &topic, Partition: partition}}, 5000)
if err != nil {
return fmt.Errorf("xxx")
}

// Assign offset to the consumer
var offset kafka.Offset
if len(lastOffsets) > 0 {
offset = lastOffsets[partition].Offset
}

err = consumer.Assign([]kafka.TopicPartition{{Topic: &topic, Partition: partition, Offset: offset}})
if err != nil {
return fmt.Errorf("xxx")
}`

We have kafka in docker provided by confluentinc/cp-kafka, confluentinc/cp-zookeeper. The kafka container is run in the following way:

docker run -d
--name=kafka
--hostname=kafka
--publish=0.0.0.0:xxxx:xxxx
-e KAFKA_ZOOKEEPER_CONNECT=x.x.x.x:xxxxx
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://x.x.x.x:xxxx
-e KAFKA_BROKER_ID=2
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
confluentinc/cp-kafka:latest

Checklist

Please provide the following information:

• confluent-kafka-go and librdkafka version (LibraryVersion()): 722175 0.11.4-12-gf26b9b
• Apache Kafka broker version:
• Client configuration: ConfigMap{...} : See above.
• Operating system: Ubuntu 16.04.4 LTS
• Provide client logs (with "debug": ".." as necessary)
• Provide broker log excerpts
• Critical issue

[rnpridgeon]

Do your consumers by chance go long periods of time without processing or committing new offsets? More specifically is it possible your consume will not commit a new offset before offsets.retention.minutes(default: 1440 (1 day) has elapsed?

I ask because even with auto offset commits librdkafka will only commit new offsets. It will not recommit the same offset which means its possible for your consumers offsets to be cleaned up by the broker. In the even this happens you will fall back on your offset reset policy.

confluentinc/librdkafka#1372

[n5i]

@rnpridgeon, very interesting. I'll dig in this direction as well. My consumer is short-lived, it lasts few seconds. I've checked the logs in the moments where the offset is reset and nothing which might cause a timeout is there. I handle all errors, but the log is clean, just offset starts from the beginning.

[edenhill]

Enable "debug": "consumer" to see what offsets are being used

[n5i]

@edenhill, I'll do, though the bug hasn't showed up for a week. From the other hand I do have offsets in my logs. It resets to -1001 and then increments as usual: 1, 2, 3, ...

[n5i]

It is duplication of #1102.
The issue is mentioned in the documentation.