kgo: fix NoAck produce requests

Kafka sends **no response** when we produce with no acks, so the current
code was wrong: we would send a request and then it would effectively
hang until the read timeout. Then we would re-send the request.

This now adds the proper shortcut logic such that as soon as we write
the request successfull, we finish all promises.

We still retry on error because if we receive an error, we did a partial
write and kafka did not understand our request.

pkg/kgo/broker.go

@@ -307,6 +307,27 @@ func (b *broker) handleReqs() {
 			continue
 		}
 
+		// Produce requests (and only produce requests) can be written
+		// without receiving a reply. If we see required acks is 0,
+		// then we immediately call the promise with no response.
+		//
+		// We provide a non-nil *kmsg.FetchResponse for
+		// *kmsg.FetchRequest just to ensure we do not return with no
+		// error and no kmsg.Response, per the client contract.
+		var isNoResp bool
+		var noResp kmsg.Response
+		switch r := req.(type) {
+		case *produceRequest:
+			isNoResp = r.acks == 0
+		case *kmsg.ProduceRequest:
+			isNoResp = r.Acks == 0
+			noResp = &kmsg.ProduceResponse{Version: req.GetVersion()}
+		}
+		if isNoResp {
+			pr.promise(noResp, nil)
+			continue
+		}
+
 		rt, _ := cxn.cl.connTimeoutFn(req)
 
 		cxn.waitResp(promisedResp{

pkg/kgo/record_and_fetch.go

@@ -122,8 +122,10 @@ type Record struct {
 
 	// Offset is the offset that a record is written as.
 	//
-	// For producing, this is left unset. This will be set by the client
-	// as appropriate.
+	// For producing, this is left unset. This will be set by the client as
+	// appropriate. If you are producing with no acks, this will just be
+	// the offset used in the produce request and does not mirror the
+	// offset actually stored within Kafka.
 	Offset int64
 }
 

pkg/kgo/sink.go

@@ -541,6 +541,22 @@ func (s *sink) handleReqResp(req *produceRequest, resp kmsg.Response, err error)
 	s.firstRespCheck(req.version)
 	atomic.StoreUint32(&s.consecutiveFailures, 0)
 
+	// If we have no acks, we will have no response. The following block is
+	// basically an extremely condensed version of everything that follows.
+	// We *do* retry on error even with no acks, because an error would
+	// mean the write itself failed.
+	if req.acks == 0 {
+		for _, partitions := range req.batches {
+			for partition, batch := range partitions {
+				if !batch.isFirstBatchInRecordBuf() {
+					continue
+				}
+				s.cl.finishBatch(batch.recBatch, req.producerID, req.producerEpoch, partition, 0, nil)
+			}
+		}
+		return
+	}
+
 	var reqRetry seqRecBatches // handled at the end
 
 	pr := resp.(*kmsg.ProduceResponse)