Handling 0 as possible required ack value and InvalidRequiredAcks errors

conn.go

@@ -870,7 +870,7 @@ func (c *Conn) WriteCompressedMessages(codec CompressionCodec, msgs ...Message)
 // connection requests when producing messages.
 func (c *Conn) SetRequiredAcks(n int) error {
 	switch n {
-	case -1, 1:
+	case -1, 0, 1:
 		atomic.StoreInt32(&c.requiredAcks, int32(n))
 		return nil
 	default:
@@ -939,21 +939,23 @@ func (c *Conn) do(d *connDeadline, write func(time.Time, int32) error, read func
 		return err
 	}
 
-	deadline, size, lock, err := c.waitResponse(d, id)
-	if err != nil {
-		return err
-	}
+	if c.requiredAcks != 0 {
+		deadline, size, lock, err := c.waitResponse(d, id)
+		if err != nil {
+			return err
+		}
 
-	if err = read(deadline, size); err != nil {
-		switch err.(type) {
-		case Error:
-		default:
-			c.conn.Close()
+		if err = read(deadline, size); err != nil {
+			switch err.(type) {
+			case Error:
+			default:
+				c.conn.Close()
+			}
 		}
+		lock.Unlock()
 	}
 
 	d.unsetConnReadDeadline()
-	lock.Unlock()
 	return err
 }
 

writer.go

@@ -636,7 +636,8 @@ func (w *writer) dial() (conn *Conn, err error) {
 			t1 := time.Now()
 			w.stats.dials.observe(1)
 			w.stats.dialTime.observeDuration(t1.Sub(t0))
-			conn.SetRequiredAcks(w.requiredAcks)
+			// set the error for a potential invalid requiredAcks value
+			err = conn.SetRequiredAcks(w.requiredAcks)
 			break
 		}
 	}