sink: bugfix sequence number wrapping for EOS

KIP-98 specifies that a sequence number will "wrap" on overflow. There
is no description of what wrapping is, so the obvious implementation is
just to wrap negative.

While browsing the Kafka source, I noticed that they actually just wrap
back to 0. So, I tested producing >2**31 records, and the first batch
after wrapping causes INVALID_RECORD problems. I then changed to this
logic (mirroring Kafka's) and producing >2**31 records is fine.

pkg/kgo/sink.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"hash/crc32"
+	"math"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -122,7 +123,11 @@ func (s *sink) createReq(id int64, epoch int16) (*produceRequest, *kmsg.AddParti
 		recBuf.inflight++
 
 		recBuf.batchDrainIdx++
-		recBuf.seq += int32(len(batch.records))
+		if recBuf.seq > math.MaxInt32-int32(len(batch.records)) {
+			recBuf.seq = int32(len(batch.records)) - (math.MaxInt32 - recBuf.seq) - 1
+		} else {
+			recBuf.seq += int32(len(batch.records))
+		}
 		moreToDrain = moreToDrain || recBuf.tryStopLingerForDraining()
 		recBuf.mu.Unlock()