test(consumer_test.go): add tests for empty record value

Author: luiz-nogueira

src/injector/store/store.go

@@ -21,6 +21,10 @@ type basicStore struct {
 }
 
 func (s basicStore) Insert(records []*models.Record) error {
+	if len(records) == 0 {
+		return nil
+	}
+	
 	elasticRecords, err := s.codec.EncodeElasticRecords(records)
 	if err != nil {
 		return err

src/kafka/consumer_test.go

@@ -139,3 +139,47 @@ func TestKafka_Start(t *testing.T) {
 	db.GetClient().DeleteByQuery(esIndex).Query(elastic.MatchAllQuery{}).Do(context.Background())
 	db.CloseClient()
 }
+
+func TestKafka_Start_WithNilValue(t *testing.T) {
+	signals := make(chan os.Signal, 1)
+	notifications := make(chan Notification, 1)
+	go k.Start(signals, notifications)
+	config := sarama.NewConfig()
+	config.Producer.Return.Successes = true
+	config.Producer.MaxMessageBytes = 20 * 1024 * 1024 // 20mb
+	config.Producer.Flush.Frequency = 1 * time.Millisecond
+	config.Version = sarama.V0_10_0_0 // This version is the same as in production
+	<-notifications
+	producer, err := fixtures.NewProducer("localhost:9092", config, schemaRegistry)
+	expectedTimestamp := time.Now().UnixNano() / int64(time.Millisecond)
+	rec := fixtures.NewEmptyFixtureRecord()
+	var msg *sarama.ProducerMessage
+	if assert.NoError(t, err) {
+
+		err = producer.Publish(rec)
+		if assert.NoError(t, err) {
+			msg = <-producer.GetSuccesses()
+		} else {
+			fmt.Println(err.Error())
+		}
+	}
+	<-notifications
+	esIndex := fmt.Sprintf("%s-%s", msg.Topic, time.Now().Format("2006-01-02"))
+	esId := fmt.Sprintf("%d:%d", msg.Partition, msg.Offset)
+	_, err = db.GetClient().Refresh(esIndex).Do(context.Background())
+	if assert.NoError(t, err) {
+		res, err := db.GetClient().Get().Index(esIndex).Id(esId).Do(context.Background())
+		var r fixtures.FixtureRecord
+		if assert.NoError(t, err) {
+			assert.True(t, res.Found)
+			err = json.Unmarshal(res.Source, &r)
+			if assert.NoError(t, err) {
+				assert.Equal(t, rec.Id, r.Id)
+				assert.InDelta(t, expectedTimestamp, r.Timestamp, 5000.0)
+			}
+		}
+		signals <- os.Interrupt
+	}
+	db.GetClient().DeleteByQuery(esIndex).Query(elastic.MatchAllQuery{}).Do(context.Background())
+	db.CloseClient()
+}

src/kafka/encode_decode_test.go

@@ -57,5 +57,5 @@ func TestDecoder_AvroMessageToRecord_NilMessageValue(t *testing.T) {
 	record, err := d.AvroMessageToRecord(nil, &sarama.ConsumerMessage{Value: nil})
 	isErrNilMessage := errors.Is(err, e.ErrNilMessage)
 	assert.Nil(t, record)
-	assert.Equal(t, isErrNilMessage, true)
+	assert.True(t, isErrNilMessage)
 }

src/kafka/fixtures/fixtures.go

@@ -38,6 +38,10 @@ func NewFixtureRecord() *FixtureRecord {
 	return &FixtureRecord{Id: rand.Int31()}
 }
 
+func NewEmptyFixtureRecord() *FixtureRecord {
+	return &FixtureRecord{}
+}
+
 func NewRecord(ts time.Time) (*models.Record, int32, int32) {
 	id := rand.Int31()
 	value := rand.Int31()