feat(errors): add ErrNilMessage

Author: luiz-nogueira

src/elasticsearch/elasticsearch.go

@@ -89,7 +89,7 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 		return &InsertResponse{AlreadyExists: nil, Retry: records, Backoff: true}, nil
 	}
 	if err != nil {
-		level.Debug(d.logger).Log("message", "something went wrong with elasticsearch", "err", err)
+		_ = level.Debug(d.logger).Log("message", "something went wrong with elasticsearch", "err", err)
 		return nil, err
 	}
 	if res.Errors {
@@ -113,19 +113,18 @@ func (d recordDatabase) Insert(records []*models.ElasticRecord) (*InsertResponse
 			}
 			for _, f := range failed {
 				if f.Status == http.StatusBadRequest {
-					level.Debug(d.logger).Log("message", "elasticsearch bad requests", "err", f)
+					_ = level.Debug(d.logger).Log("message", "elasticsearch bad requests", "err", f)
 					d.metricsPublisher.ElasticsearchBadRequests(1)
 					continue
 				}
 				if f.Status == http.StatusConflict {
-					level.Debug(d.logger).Log("message", "elasticsearch conflicts", "err", f)
+					_ = level.Debug(d.logger).Log("message", "elasticsearch conflicts", "err", f)
 					d.metricsPublisher.ElasticsearchConflicts(1)
 					continue
 				}
 				retry = append(retry, recordMap[f.Id])
 				if f.Status == http.StatusTooManyRequests {
 					//es is overloaded, backoff
-					level.Debug(d.logger).Log("message", "es is overloaded, backoff")
 					overloaded = true
 				}
 			}

src/errors/ErrNilMessage.go

@@ -0,0 +1,7 @@
+package errors
+
+type ErrNilMessage struct {}
+
+func (err *ErrNilMessage) Error() string {
+	return "message value is nil"
+}

src/kafka/consumer.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"os"
+	"errors"
 
 	"time"
 
@@ -13,6 +14,7 @@ import (
 	"github.com/go-kit/kit/log/level"
 	"github.com/inloco/kafka-elasticsearch-injector/src/metrics"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
+	e "github.com/inloco/kafka-elasticsearch-injector/src/errors"
 )
 
 type Notification int32
@@ -144,6 +146,10 @@ func (k *kafka) worker(consumer *cluster.Consumer, buffSize int, notifications c
 				for _, msg := range buf {
 					req, err := k.consumer.Decoder(nil, msg)
 					if err != nil {
+						if errors.Is(err, e.ErrNilMessage) {
+							continue
+						}
+
 						level.Error(k.consumer.Logger).Log(
 							"message", "Error decoding message",
 							"err", err.Error(),

src/kafka/encode_decode.go

@@ -13,6 +13,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/linkedin/goavro/v2"
 	"github.com/inloco/kafka-elasticsearch-injector/src/models"
+	e "github.com/inloco/kafka-elasticsearch-injector/src/errors"
 	"github.com/inloco/kafka-elasticsearch-injector/src/schema_registry"
 )
 
@@ -39,13 +40,7 @@ func (d *Decoder) DeserializerFor(recordType string) DecodeMessageFunc {
 
 func (d *Decoder) AvroMessageToRecord(context context.Context, msg *sarama.ConsumerMessage) (*models.Record, error) {
 	if msg.Value == nil {
-		return &models.Record{
-			Topic:     msg.Topic,
-			Partition: msg.Partition,
-			Offset:    msg.Offset,
-			Timestamp: msg.Timestamp,
-			Json:      nil,
-		}, nil
+		return nil, new(e.ErrNilMessage)
 	}
 
 	schemaId := getSchemaId(msg)

src/kafka/encode_decode_test.go

@@ -6,9 +6,11 @@ import (
 	"sync"
 	"testing"
 	"time"
+	"errors"
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
+	e "github.com/inloco/kafka-elasticsearch-injector/src/errors"
 )
 
 type dummy struct {
@@ -49,3 +51,11 @@ func TestDecoder_JsonMessageToRecord_MalformedJson(t *testing.T) {
 	assert.Nil(t, record)
 	assert.NotNil(t, err)
 }
+
+func TestDecoder_AvroMessageToRecord_NilMessageValue(t *testing.T) {
+	d := &Decoder{CodecCache: sync.Map{}}
+	record, err := d.AvroMessageToRecord(nil, &sarama.ConsumerMessage{Value: nil})
+	isErrNilMessage := errors.Is(err, e.ErrNilMessage)
+	assert.Nil(t, record)
+	assert.Equal(t, isErrNilMessage, true)
+}