Fix for panic when RequiredAcks is set to RequireNone (#504)

* Fix panic in async wait() method when RequiredAcks is None

When RequiredAcks is None, the producer does not wait for a
response from the broker, therefore the response is nil.
The async wait() method was not handling this case, leading
to a panic.

* Add regression test for RequiredAcks == RequireNone

This new test is required because all the other Writer tests use
NewWriter() to create Writers, which sets RequiredAcks to
RequireAll when 0 (None) was specified.

Author: neilcook

transport.go

@@ -778,6 +778,8 @@ func (p async) await(ctx context.Context) (Response, error) {
 	select {
 	case x := <-p:
 		switch v := x.(type) {
+		case nil:
+			return nil, nil // A nil response is ok (e.g. when RequiredAcks is None)
 		case Response:
 			return v, nil
 		case error:

writer_test.go

@@ -51,6 +51,10 @@ func TestWriter(t *testing.T) {
 			scenario: "setting a non default balancer on the writer",
 			function: testWriterSetsRightBalancer,
 		},
+		{
+			scenario: "setting RequiredAcks to None in Writer doesn't cause a panic",
+			function: testWriterRequiredAcksNone,
+		},
 	}
 
 	for _, test := range tests {
@@ -83,6 +87,29 @@ func testWriterClose(t *testing.T) {
 	}
 }
 
+func testWriterRequiredAcksNone(t *testing.T) {
+	topic := makeTopic()
+	createTopic(t, topic, 1)
+	defer deleteTopic(t, topic)
+
+	w := &Writer{
+		Addr:     TCP("localhost:9092"),
+		Topic:    topic,
+		Balancer: &RoundRobin{},
+		RequiredAcks: RequireNone,
+	}
+	defer w.Close()
+
+	msg := Message{
+		Key:   []byte("ThisIsAKey"),
+		Value: []byte("Test message for required acks test")}
+
+	err := w.WriteMessages(context.Background(), msg)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
 func testWriterSetsRightBalancer(t *testing.T) {
 	const topic = "test-writer-1"
 	balancer := &CRC32Balancer{}