adds a function to get consumer handle

kafka/chandle.go

@@ -0,0 +1,17 @@
+package kafka
+
+import (
+	"errors"
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+)
+
+func ConsumerHandle(group *ConsumerGroup) (*kafka.Consumer, error) {
+	if group.c == nil {
+		return nil, errors.New("consumer handle not initialized")
+	}
+	kc, ok := group.c.(*kafka.Consumer)
+	if !ok {
+		return nil, errors.New("could not assert to *kafka.Consumer")
+	}
+	return kc, nil
+}

kafka/consumer.go

@@ -17,6 +17,7 @@ type ConsumerGroup struct {
 	Logger           ziggurat.StructuredLogger
 	GroupConfig      ConsumerConfig
 	wg               *sync.WaitGroup
+	c                confluentConsumer
 	consumerMakeFunc func(*kafka.ConfigMap, []string) confluentConsumer
 }
 
@@ -34,6 +35,7 @@ func (cg *ConsumerGroup) Consume(ctx context.Context, handler ziggurat.Handler)
 
 	cm := cg.GroupConfig.toConfigMap()
 	confCons := cg.consumerMakeFunc(&cm, cg.GroupConfig.Topics)
+	cg.c = confCons
 	for i := 0; i < consConf.ConsumerCount; i++ {
 		workerID := fmt.Sprintf("%s_%d", groupID, i)
 		w := worker{