feat(kafka): add ConsumerFactory in order to align future developments (#192)

Author: smoya

kafka/consume.go

@@ -14,6 +14,11 @@ import (
 // ErrSeekTimedOut is the error returned when a consumer timed out during Seek.
 var ErrSeekTimedOut = errors.New("Kafka Seek timed out. Please try again.")
 
+// ConsumerFactory creates a Consumer.
+// You can use it for postponing the creation of a consumer on runtime. E.g. setting an initial offset that you don't know on boot time.
+// NewConsumer and NewDetachedConsumer implements this.
+type ConsumerFactory func(logrus.FieldLogger, Config, ...ConfigOpt) (Consumer, error)
+
 // Consumer reads messages from Kafka.
 type Consumer interface {
 	// AssignPartittionByKey sets the current consumer to read from a partion by a hashed key.
@@ -62,7 +67,7 @@ type ConfluentConsumer struct {
 //       permission on the group coordinator for managing commits, so it needs a consumer group in the broker.
 //       In order to simplify, the default consumer group id is copied from the configured topic name, so make sure you have a
 //       policy that gives permission to such consumer group.
-func NewDetachedConsumer(log logrus.FieldLogger, conf Config, opts ...ConfigOpt) (*ConfluentConsumer, error) {
+func NewDetachedConsumer(log logrus.FieldLogger, conf Config, opts ...ConfigOpt) (Consumer, error) {
 	// See Reference at https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 	kafkaConf := conf.baseKafkaConfig()
 	_ = kafkaConf.SetKey("enable.auto.offset.store", false) // manually StoreOffset after processing a message. It is mandatory for detached consumers.
@@ -128,7 +133,7 @@ func NewDetachedConsumer(log logrus.FieldLogger, conf Config, opts ...ConfigOpt)
 // NewConsumer creates a ConfluentConsumer based on config.
 // - NOTE if the partition is set and the partition key is not set in config we have no way
 //   of knowing where to assign the consumer to in the case of a rebalance
-func NewConsumer(log logrus.FieldLogger, conf Config, opts ...ConfigOpt) (*ConfluentConsumer, error) {
+func NewConsumer(log logrus.FieldLogger, conf Config, opts ...ConfigOpt) (Consumer, error) {
 	// See Reference at https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
 	kafkaConf := conf.baseKafkaConfig()
 	_ = kafkaConf.SetKey("enable.auto.offset.store", false) // manually StoreOffset after processing a message. Otherwise races may happen.)

kafka/consume_test.go

@@ -22,11 +22,11 @@ func TestConsumerFetchMessageContextAwareness(t *testing.T) {
 func TestConsumerSeek(t *testing.T) {
 	c, conf := consumer(t)
 	defer checkClose(t, c)
-	require.NoError(t, c.c.Assign(kafkalib.TopicPartitions{{Topic: &conf.Topic, Partition: 0}})) // manually assign partition
+	require.NoError(t, c.(*ConfluentConsumer).c.Assign(kafkalib.TopicPartitions{{Topic: &conf.Topic, Partition: 0}})) // manually assign partition
 	require.NoError(t, c.Seek(2))
 }
 
-func consumer(t *testing.T) (*ConfluentConsumer, Config) {
+func consumer(t *testing.T) (Consumer, Config) {
 	conf := Config{
 		Brokers: nil, // No brokers are used for unit test.
 		Topic:   "gotest",

kafka/kafkatest/utils.go

@@ -12,6 +12,18 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+func FakeKafkaConsumerFactory(distri <-chan *kafkalib.Message) kafka.ConsumerFactory {
+	return func(log logrus.FieldLogger, _ kafka.Config, _ ...kafka.ConfigOpt) (kafka.Consumer, error) {
+		return NewFakeKafkaConsumer(log, distri), nil
+	}
+}
+
+func KafkaConsumerFactoryFromConsumer(c kafka.Consumer) kafka.ConsumerFactory {
+	return func(_ logrus.FieldLogger, _ kafka.Config, _ ...kafka.ConfigOpt) (kafka.Consumer, error) {
+		return c, nil
+	}
+}
+
 func KafkaPipe(log logrus.FieldLogger) (*FakeKafkaConsumer, *FakeKafkaProducer) {
 	distri := make(chan *kafkalib.Message, 200)
 	rdr := NewFakeKafkaConsumer(log, distri)