feat(injector): add WithKeyAndValue variable

Author: luiz-nogueira

cmd/injector.go

@@ -45,6 +45,7 @@ func main() {
 		BufferSize:            os.Getenv("KAFKA_CONSUMER_BUFFER_SIZE"),
 		MetricsUpdateInterval: os.Getenv("KAFKA_CONSUMER_METRICS_UPDATE_INTERVAL"),
 		RecordType:            os.Getenv("KAFKA_CONSUMER_RECORD_TYPE"),
+		WithKeyAndValue:       os.Getenv("KAFKA_CONSUMER_WITH_KEY_AND_VALUE"),
 	}
 	metricsPublisher := metrics.NewMetricsPublisher()
 	service := injector.NewService(logger, metricsPublisher)

src/injector/injector.go

@@ -37,6 +37,12 @@ func MakeKafkaConsumer(endpoints Endpoints, logger log.Logger, schemaRegistry *s
 		SchemaRegistry: schemaRegistry,
 	}
 
+	withKeyAndValue, err := strconv.ParseBool(kafkaConfig.WithKeyAndValue)
+	if err != nil {
+		level.Warn(logger).Log("err", err, "message", "failed to get consumer with key and value")
+		withKeyAndValue = false
+	}
+
 	return kafka.Consumer{
 		Topics:                kafkaConfig.Topics,
 		Group:                 kafkaConfig.ConsumerGroup,
@@ -47,5 +53,6 @@ func MakeKafkaConsumer(endpoints Endpoints, logger log.Logger, schemaRegistry *s
 		BatchSize:             batchSize,
 		MetricsUpdateInterval: metricsUpdateInterval,
 		BufferSize:            bufferSize,
+		WithKeyAndValue:       withKeyAndValue,
 	}, nil
 }

src/kafka/config.go

@@ -13,4 +13,5 @@ type Config struct {
 	MetricsUpdateInterval string
 	BufferSize            string
 	RecordType            string
+	WithKeyAndValue       string
 }

src/kafka/consumer.go

@@ -43,6 +43,7 @@ type Consumer struct {
 	BatchSize             int
 	MetricsUpdateInterval time.Duration
 	BufferSize            int
+	WithKeyAndValue       bool
 }
 
 type topicPartitionOffset struct {
@@ -144,7 +145,7 @@ func (k *kafka) worker(consumer *cluster.Consumer, buffSize int, notifications c
 		for idx == buffSize {
 			if decoded == nil {
 				for _, msg := range buf {
-					req, err := k.consumer.Decoder(nil, msg)
+					req, err := k.consumer.Decoder(nil, msg, k.consumer.WithKeyAndValue)
 					if err != nil {
 						if errors.Is(err, e.ErrNilMessage) {
 							continue

src/kafka/encode_decode.go

@@ -21,7 +21,7 @@ import (
 // message object. It's designed to be used in Kafka consumers.
 // One straightforward DecodeMessageFunc could be something that
 // Avro decodes the message body to the concrete response type.
-type DecodeMessageFunc func(context.Context, *sarama.ConsumerMessage) (record *models.Record, err error)
+type DecodeMessageFunc func(context.Context, *sarama.ConsumerMessage, bool) (record *models.Record, err error)
 
 const kafkaTimestampKey = "@timestamp"
 const keyField = "key"
@@ -39,7 +39,7 @@ func (d *Decoder) DeserializerFor(recordType string) DecodeMessageFunc {
 	}
 }
 
-func (d *Decoder) AvroMessageToRecord(context context.Context, msg *sarama.ConsumerMessage) (*models.Record, error) {
+func (d *Decoder) AvroMessageToRecord(context context.Context, msg *sarama.ConsumerMessage, withKeyAndValue bool) (*models.Record, error) {
 	if msg.Value == nil {
 		return nil, e.ErrNilMessage
 	}
@@ -63,7 +63,7 @@ func (d *Decoder) AvroMessageToRecord(context context.Context, msg *sarama.Consu
 
 	parsedNative[kafkaTimestampKey] = makeTimestamp(msg.Timestamp)
 
-	if msg.Key != nil {
+	if withKeyAndValue && msg.Key != nil {
 		nativeKey, err := d.nativeFromBinary(msg.Key)
 		if err != nil {
 			return nil, err
@@ -84,7 +84,7 @@ func makeTimestamp(timestamp time.Time) int64 {
 	return timestamp.UnixNano() / int64(time.Millisecond)
 }
 
-func (d *Decoder) JsonMessageToRecord(context context.Context, msg *sarama.ConsumerMessage) (*models.Record, error) {
+func (d *Decoder) JsonMessageToRecord(context context.Context, msg *sarama.ConsumerMessage, withKeyAndValue bool) (*models.Record, error) {
 	var jsonValue map[string]interface{}
 	err := json.Unmarshal(msg.Value, &jsonValue)
 

src/kafka/encode_decode_test.go

@@ -28,7 +28,7 @@ func TestDecoder_JsonMessageToRecord(t *testing.T) {
 		Partition: 1,
 		Offset:    54,
 		Timestamp: time.Now(),
-	})
+	}, false)
 	assert.Nil(t, err)
 	returnedJsonBytes, err := json.Marshal(record.Json)
 	assert.Nil(t, err)
@@ -47,7 +47,7 @@ func TestDecoder_JsonMessageToRecord_MalformedJson(t *testing.T) {
 		Partition: 1,
 		Offset:    54,
 		Timestamp: time.Now(),
-	})
+	}, false)
 	assert.Nil(t, record)
 	assert.NotNil(t, err)
 }