feat: allow for asynchronous kafka offset commits (#236)

Author: cbosss

kafka/consume.go

@@ -36,6 +36,11 @@ type Consumer interface {
 	// CommitMessage commits the offset of a given message.
 	CommitMessage(msg *kafkalib.Message) error
 
+	// StoreOffset stores the offset of a given message. The offset will be asynchronously flushed to kafka every
+	// `auto.commit.interval.ms`. This method is non-blocking and will be faster than `CommitMessage`, however it has
+	// weaker delivery guarantees.
+	StoreOffset(msg *kafkalib.Message) error
+
 	// GetMetadata gets the metadata for a consumer.
 	GetMetadata(allTopics bool) (*kafkalib.Metadata, error)
 
@@ -337,6 +342,20 @@ func (cc *ConfluentConsumer) CommitMessage(msg *kafkalib.Message) error {
 	return errors.Wrap(err, "failed committing Kafka message")
 }
 
+// StoreOffset stores the offset of a given message. The offset will be asynchronously flushed to kafka every
+// `auto.commit.interval.ms`. This method is non-blocking and will be faster than `CommitMessage`, however it has
+// weaker delivery guarantees.
+func (cc *ConfluentConsumer) StoreOffset(msg *kafkalib.Message) error {
+	if msg.TopicPartition.Error != nil {
+		return errors.New("can't commit errored message")
+	}
+
+	offsets := []kafkalib.TopicPartition{msg.TopicPartition}
+	offsets[0].Offset++
+	_, err := cc.c.StoreOffsets(offsets)
+	return err
+}
+
 // Pause pauses consumption of the provided partitions
 func (cc *ConfluentConsumer) Pause(p []kafkalib.TopicPartition) error {
 	err := cc.c.Pause(p)

kafka/kafkatest/utils.go

@@ -121,6 +121,10 @@ func (f *FakeKafkaConsumer) CommitMessage(msg *kafkalib.Message) error {
 	return nil
 }
 
+func (f *FakeKafkaConsumer) StoreOffset(msg *kafkalib.Message) error {
+	return f.CommitMessage(msg)
+}
+
 func (f *FakeKafkaConsumer) SetInitialOffset(offset int64) error {
 	f.msgMu.Lock()
 	f.offset = offset