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