Feat: manual commit kafka ConsumeOperator implemented

Author: asheswook

internal/queue/kafka/consumer.go

@@ -48,3 +48,50 @@ func (c *Consumer) Cleanup(session sarama.ConsumerGroupSession) error {
 func (c *Consumer) Callback() func(*sarama.ConsumerMessage) {
 	return c.callback
 }
+
+type MarkableConsumer struct {
+	callback func(*sarama.ConsumerMessage, func())
+}
+
+func NewMarkableConsumer(callback func(message *sarama.ConsumerMessage, ack func())) *MarkableConsumer {
+	return &MarkableConsumer{
+		callback: callback,
+	}
+}
+
+// ConsumeClaim must start a Consumer loop of ConsumerGroupClaim's Messages().
+// NOTE: This must not be called within a goroutine, already handled as goroutines by sarama.
+func (c *MarkableConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for {
+		select {
+		case msg, ok := <-claim.Messages():
+			if !ok {
+				// 메세지 채널 닫혔을 때
+				return nil
+			}
+
+			c.callback(msg, func() {
+				session.MarkMessage(msg, "")
+			})
+
+		case <-session.Context().Done():
+			return nil
+		}
+	}
+}
+
+// Setup is run at the beginning of a new session, before ConsumeClaim.
+func (c *MarkableConsumer) Setup(session sarama.ConsumerGroupSession) error {
+	log.Print("Consumer up and running")
+	return nil
+}
+
+// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited.
+func (c *MarkableConsumer) Cleanup(session sarama.ConsumerGroupSession) error {
+	log.Print("Consumer stopped")
+	return nil
+}
+
+func (c *MarkableConsumer) Callback() func(*sarama.ConsumerMessage, func()) {
+	return c.callback
+}

internal/queue/kafka/operator.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"github.com/IBM/sarama"
 	"github.com/violetpay-org/event-queue/queue"
+	"log"
 )
 
 type ConsumeOperator[Msg any] struct {
@@ -100,6 +101,148 @@ func (k *ConsumeOperator[Msg]) StopConsume() error {
 	return nil
 }
 
+// AckConsumeOperator 는 Acknowledge (메세지 브로커에게 보내는 메세지 정상 수신 응답) 을 다루는 컨슈머입니다. 실행되는 흐름은 다음과 같습니다.
+//
+// 1. 내부 컨슈머가 메세지를 받습니다.
+//
+// 2. 애크 콜백 함수 (AckCallback) 를 호출합니다.
+// 만약 AckCallback 함수가 true 를 반환하면, 메세지를 정상적으로 수신했음을 브로커에 알립니다. 이후 다음 과정으로 진행합니다.
+// 만약 AckCallback 함수가 false 를 반환하면, 메세지를 수신했음을 브로커에 알리지 않습니다. 이후 다음 과정으로 진행하지 않고 즉시 처리 과정을 종료합니다.
+//
+// 3. 콜백 함수 (Callback) 를 호출합니다.
+//
+// 4. 종료합니다.
+type AckConsumeOperator[Msg any] struct {
+	serializer queue.MessageSerializer[*sarama.ConsumerMessage, Msg]
+
+	ackCallback queue.AckCallback[Msg]
+	callback    queue.Callback[Msg]
+
+	initialized bool
+	cancel      *context.CancelFunc
+
+	consumer *MarkableConsumer
+	brokers  []string
+	topic    string
+	groupId  string
+	config   *sarama.Config
+}
+
+func NewAckConsumeOperator[Msg any](serializer queue.MessageSerializer[*sarama.ConsumerMessage, Msg], ackCallback queue.AckCallback[Msg], callback queue.Callback[Msg], brokers []string, topic string, groupId string, config *sarama.Config) *AckConsumeOperator[Msg] {
+	return &AckConsumeOperator[Msg]{
+		serializer:  serializer,
+		ackCallback: ackCallback,
+		callback:    callback,
+		brokers:     brokers,
+		topic:       topic,
+		groupId:     groupId,
+		config:      config,
+	}
+}
+
+func (k *AckConsumeOperator[Msg]) QueueName() string {
+	return k.topic
+}
+
+func (k *AckConsumeOperator[Msg]) Serializer() queue.MessageSerializer[*sarama.ConsumerMessage, Msg] {
+	return k.serializer
+}
+
+func (k *AckConsumeOperator[Msg]) AckCallback() queue.AckCallback[Msg] {
+	return k.ackCallback
+}
+
+func (k *AckConsumeOperator[Msg]) Callback() queue.Callback[Msg] {
+	return func(msg Msg) {
+		k.callback(msg)
+	}
+}
+
+// BeforeConsume 은 Serializer 를 호출한 후 AckCallback 함수를 실행하는 함수입니다.
+// 실제 컨슘할 때에는 AckCallback 결과에 따라 Callback 함수를 실행할지 결정합니다.
+func (k *AckConsumeOperator[Msg]) BeforeConsume(msg *sarama.ConsumerMessage, ack func()) (sMsg Msg, ok bool) {
+	var err error
+	sMsg, err = k.serializer.Serialize(msg)
+	if err != nil {
+		log.Fatal("Serializing error: ", err)
+		return sMsg, false
+	}
+
+	if !k.ackCallback(sMsg) {
+		return sMsg, false
+	}
+
+	ack()
+	return sMsg, true
+}
+
+// Consume 은 Serializer 를 호출한 후 Callback 함수를 실행하는 함수입니다.
+// 주의: 실제 컨슘할 때의 로직과는 다릅니다. 실제로는 BeforeConsume 함수를 통해 AckCallback 함수를 실행한 뒤 Ack 결과에 따라 Callback 을 실행합니다.
+// ConsumeOperator 를 구현함과 동시에 테스트 가능성을 높이기 위해 이 함수를 따로 분리했습니다.
+func (k *AckConsumeOperator[Msg]) Consume(msg *sarama.ConsumerMessage) {
+	sMsg, err := k.serializer.Serialize(msg)
+	if err != nil {
+		return
+	}
+
+	k.callback(sMsg)
+}
+
+func (k *AckConsumeOperator[Msg]) Init() {
+	consumer := NewMarkableConsumer(func(msg *sarama.ConsumerMessage, ack func()) {
+		sMsg, ok := k.BeforeConsume(msg, ack)
+		if !ok {
+			return
+		}
+
+		// goroutine 으로 실행! BeforeConsume 이 완료되면 Consume된 상태이므로 goroutine으로 실행합니다..
+		go k.callback(sMsg)
+	})
+
+	k.consumer = consumer
+	k.initialized = true
+}
+
+func (k *AckConsumeOperator[Msg]) StartConsume() error {
+	if !k.initialized {
+		k.Init()
+	}
+
+	client, err := sarama.NewConsumerGroup(k.brokers, k.groupId, k.config)
+	if err != nil {
+		return err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	k.cancel = &cancel
+
+	go func() {
+		for {
+			if err := client.Consume(ctx, []string{k.topic}, k.consumer); err != nil {
+				if errors.Is(err, sarama.ErrClosedConsumerGroup) {
+					return
+				}
+			}
+
+			if ctx.Err() != nil {
+				return
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (k *AckConsumeOperator[Msg]) StopConsume() error {
+	if k.cancel == nil {
+		return errors.New("tried to stop consume before starting (or operator no started, but StopConsume called)")
+	}
+
+	(*k.cancel)()
+
+	return nil
+}
+
 type BytesProduceOperatorCtor struct {
 	pool *ProducerPool
 

internal/queue/kafka/operator_test.go

@@ -12,6 +12,13 @@ func mapConsumeOperatorProvider(queueName string, serializer queue.MessageSerial
 	return NewConsumeOperator(serializer, callback, brokers, queueName, "test-group-id", sarama.NewConfig())
 }
 
+func mapAckConsumeOperatorProvider(queueName string, serializer queue.MessageSerializer[*sarama.ConsumerMessage, map[string]string], callback queue.Callback[map[string]string]) queue.ConsumeOperator[*sarama.ConsumerMessage, map[string]string] {
+	f := func(_ map[string]string) bool {
+		return true
+	}
+	return NewAckConsumeOperator(serializer, f, callback, brokers, queueName, "test-group-id", sarama.NewConfig())
+}
+
 func consumeMessageProvider() *sarama.ConsumerMessage {
 	return &sarama.ConsumerMessage{}
 }
@@ -23,7 +30,14 @@ func TestConsumeOperator(t *testing.T) {
 			mapConsumeOperatorProvider,
 			consumeMessageProvider,
 		)
+
+		queue.TestSuiteConsumeOperator[*sarama.ConsumerMessage, map[string]string](
+			t,
+			mapAckConsumeOperatorProvider,
+			consumeMessageProvider,
+		)
 	})
+
 }
 
 func bytesProduceOperatorProvider(queueName string) queue.ProduceOperator[[]byte] {

queue/callback.go

@@ -1,3 +1,17 @@
 package queue
 
+// Callback 는 메시지를 처리하는 콜백 함수입니다.
 type Callback[M any] func(req M)
+
+// AckCallback 는 Acknowledge 를 처리하는 콜백 함수 타입입니다. 실행되는 흐름은 아래와 같습니다.
+//
+// 1. 내부 컨슈머가 메세지를 받습니다.
+//
+// 2. 애크 콜백 함수 (AckCallback) 를 호출합니다.
+// 만약 AckCallback 함수가 true 를 반환하면, 메세지를 정상적으로 수신했음을 브로커에 알립니다. 이후 다음 과정으로 진행합니다.
+// 만약 AckCallback 함수가 false 를 반환하면, 메세지를 수신했음을 브로커에 알리지 않습니다. 이후 다음 과정으로 진행하지 않고 즉시 처리 과정을 종료합니다.
+//
+// 3. 콜백 함수 (Callback) 를 호출합니다.
+//
+// 4. 종료합니다.
+type AckCallback[M any] func(req M) bool

queue/kafka/operator.go

@@ -10,6 +10,10 @@ func NewConsumeOperator[Msg any](serializer queue.MessageSerializer[*sarama.Cons
 	return kafkaqueue.NewConsumeOperator(serializer, callback, brokers, topic, groupId, config)
 }
 
+func NewAckConsumeOperator[Msg any](serializer queue.MessageSerializer[*sarama.ConsumerMessage, Msg], ackCallback queue.AckCallback[Msg], callback queue.Callback[Msg], brokers []string, topic string, groupId string, config *sarama.Config) *kafkaqueue.AckConsumeOperator[Msg] {
+	return kafkaqueue.NewAckConsumeOperator(serializer, ackCallback, callback, brokers, topic, groupId, config)
+}
+
 func NewProduceOperatorCtor(brokers []string, configProvider func() *sarama.Config) *kafkaqueue.BytesProduceOperatorCtor {
 	return kafkaqueue.NewBytesProduceOperatorCtor(brokers, configProvider)
 }