fixes consumer commit loss

CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres
 to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [v2.0.20] 2024-03-25
+
+- Manually commit uncommitted offsets before closing the Kafka Consumer
+
 ## [v2.0.18] 2024-03-25
 
 # Removed

cmd/ziggurat/main.go

@@ -20,7 +20,7 @@ func die(err error) {
 
 //go:embed templates/*
 var res embed.FS
-var Version = "v2.0.18"
+var Version = "v2.0.20"
 
 var definePaths = func(basePath string) map[string]string {
 	return map[string]string{

kafka/confluent.go

@@ -10,6 +10,7 @@ type confluentConsumer interface {
 	Poll(int) kafka.Event
 	StoreOffsets([]kafka.TopicPartition) ([]kafka.TopicPartition, error)
 	Logs() chan kafka.LogEvent
+	Commit() ([]kafka.TopicPartition, error)
 	Close() error
 }
 
@@ -28,6 +29,11 @@ func (m *MockConsumer) StoreOffsets(partitions []kafka.TopicPartition) ([]kafka.
 	return args.Get(0).([]kafka.TopicPartition), args.Error(1)
 }
 
+func (m *MockConsumer) Commit() ([]kafka.TopicPartition, error) {
+	args := m.Called()
+	return args.Get(0).([]kafka.TopicPartition), args.Error(1)
+}
+
 func (m *MockConsumer) Close() error {
 	return m.Called().Error(0)
 }

kafka/consumer_test.go

@@ -36,6 +36,7 @@ func TestWorkerOrchestration(t *testing.T) {
 		mc.On("StoreOffsets", []kafka.TopicPartition{expectedTopicPartStoreOffsets}).
 			Return([]kafka.TopicPartition{expectedTopicPartStoreOffsets}, nil)
 		mc.On("Close").Return(nil)
+		mc.On("Commit").Return([]kafka.TopicPartition{}, nil)
 		mc.On("Logs").Return(logChan)
 
 		var msgCount int32

kafka/worker.go

@@ -30,7 +30,11 @@ type worker struct {
 func (w *worker) run(ctx context.Context) {
 
 	defer func() {
-		err := w.consumer.Close()
+		_, err := w.consumer.Commit()
+		if err != nil {
+			w.logger.Error("pre-close commit error", err, map[string]interface{}{"Worker-ID": w.id})
+		}
+		err = w.consumer.Close()
 		w.logger.Error("error closing kafka consumer", err, map[string]interface{}{"Worker-ID": w.id})
 	}()
 

kafka/worker_test.go

@@ -61,6 +61,7 @@ func TestWorker(t *testing.T) {
 		mc.On("Poll", 100).Return(&kafka.Message{
 			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: 1},
 		})
+		mc.On("Commit").Return([]kafka.TopicPartition{}, nil)
 		mc.On("StoreOffsets", mock.AnythingOfType("[]kafka.TopicPartition")).Return([]kafka.TopicPartition{}, nil)
 		mc.On("Close").Return(nil)
 		ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
@@ -89,6 +90,7 @@ func TestWorker(t *testing.T) {
 
 		mc.On("Logs").Return(make(chan kafka.LogEvent))
 		mc.On("Poll", 100).Return(kafka.NewError(kafka.ErrAllBrokersDown, "fatal error", true))
+		mc.On("Commit").Return([]kafka.TopicPartition{}, nil)
 		mc.On("Close").Return(nil)
 		ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
 		defer cancel()
@@ -116,6 +118,7 @@ func TestWorker(t *testing.T) {
 
 		topic := "foo"
 		mc.On("Logs").Return(make(chan kafka.LogEvent))
+		mc.On("Commit").Return([]kafka.TopicPartition{}, nil)
 		mc.On("StoreOffsets", mock.AnythingOfType("[]kafka.TopicPartition")).Return([]kafka.TopicPartition{}, nil)
 		mc.On("Poll", 100).Return(&kafka.Message{
 			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: 1},