From d5c804739cf4e72fb3d8cdf4e2b5438b109b6e50 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Enrique=20J=2E=20Hern=C3=A1ndez?=
 <sixstone-qq@users.noreply.github.com>
Date: Mon, 4 Apr 2022 23:11:49 +0200
Subject: [PATCH] Make null value to fail to produce

To avoid producing null value messages
---
 produce.go      | 16 ++++++++++------
 produce_test.go | 14 ++++++++++++++
 2 files changed, 24 insertions(+), 6 deletions(-)

diff --git a/produce.go b/produce.go
index 418991b..0be1d4b 100644
--- a/produce.go
+++ b/produce.go
@@ -200,13 +200,17 @@ func (cmd *produceCmd) makeSaramaMessage(msg producerMessage) (*sarama.ProducerM
 		}
 		sm.Key = sarama.ByteEncoder(key)
 	}
-	if msg.Value != nil {
-		value, err := cmd.decodeValue(msg.Value)
-		if err != nil {
-			return nil, fmt.Errorf("cannot decode value: %v", err)
-		}
-		sm.Value = sarama.ByteEncoder(value)
+
+	if msg.Value == nil {
+		return nil, fmt.Errorf(`empty "value" JSON key to produce: %+v`, msg)
 	}
+
+	value, err := cmd.decodeValue(msg.Value)
+	if err != nil {
+		return nil, fmt.Errorf("cannot decode value: %v", err)
+	}
+	sm.Value = sarama.ByteEncoder(value)
+
 	if msg.Timestamp != nil {
 		sm.Timestamp = *msg.Timestamp
 	}
diff --git a/produce_test.go b/produce_test.go
index 552061a..f68349d 100644
--- a/produce_test.go
+++ b/produce_test.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/Shopify/sarama"
 	qt "github.com/frankban/quicktest"
@@ -128,6 +129,19 @@ func TestMakeSaramaMessage(t *testing.T) {
 		c.Assert(err, qt.IsNil)
 		c.Assert(gotValue, qt.ContentEquals, expectedBlob)
 	})
+
+	c.Run("Null value should fail", func(c *qt.C) {
+		cmd := new(produceCmd)
+		cmd.decodeKey, err = cmd.decoderForType("key", "string")
+		c.Assert(err, qt.IsNil)
+		pn := int32(1)
+		for _, pmsg := range []producerMessage{{}, {Key: json.RawMessage(`"k"`)}, {Partition: &pn}, {Timestamp: &time.Time{}}} {
+			msg, err := cmd.makeSaramaMessage(pmsg)
+			c.Assert(err, qt.IsNotNil)
+			c.Assert(err, qt.ErrorMatches, `empty "value" JSON key to produce:.*`)
+			c.Assert(msg, qt.IsNil)
+		}
+	})
 }
 
 func TestDeserializeLines(t *testing.T) {