Merge pull request #44 from heetch/sixstone-001-fix-empty-msg-produced

Make null value to fail to produce

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
 	}

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) {