producing: set Attrs properly before calling produce callback

The only missing field now is LeaderEpoch; setting that would involve
more guts changes than really are worth it at the moment. Currently, the
producer does not need to know the leader epoch when producing.

pkg/kgo/sink.go

@@ -753,6 +753,14 @@ func (cl *Client) finishBatch(batch *recBatch, producerID int64, producerEpoch i
 		pnr.Partition = partition
 		pnr.ProducerID = producerID
 		pnr.ProducerEpoch = producerEpoch
+
+		// A recBuf.attrs is updated when appending to be written.  For
+		// v0 && v1 produce requests, we set bit 8 in the attrs
+		// corresponding to our own RecordAttr's bit 8 being no
+		// timestamp type. Thus, we can directly convert the batch
+		// attrs to our own RecordAttrs.
+		pnr.Attrs = RecordAttrs{uint8(batch.attrs)}
+
 		cl.finishRecordPromise(pnr.promisedRec, err)
 		batch.records[i] = noPNR
 	}
@@ -1244,7 +1252,7 @@ type recBatch struct {
 	v1wireLength int32 // same as wireLength, but for message set v1
 	wireLength   int32 // tracks total size this batch would currently encode as
 
-	attrs          int16
+	attrs          int16 // updated during apending; read and converted to RecordAttrs on success
 	firstTimestamp int64 // since unix epoch, in millis
 
 	// mu guards against records being concurrently modified in
@@ -1504,11 +1512,11 @@ func (r seqRecBatch) appendTo(
 	dst = kbin.AppendInt32(dst, 0) // reserved crc
 
 	attrsAt := len(dst) // in case compression adjusting
+	r.attrs = 0
 	if transactional {
 		r.attrs |= 0x0010 // bit 5 is the "is transactional" bit
 	}
-	attrs := r.attrs
-	dst = kbin.AppendInt16(dst, attrs)
+	dst = kbin.AppendInt16(dst, r.attrs)
 	dst = kbin.AppendInt32(dst, int32(len(r.records)-1)) // lastOffsetDelta
 	dst = kbin.AppendInt64(dst, r.firstTimestamp)
 
@@ -1543,10 +1551,10 @@ func (r seqRecBatch) appendTo(
 			savings := int32(len(toCompress) - len(compressed))
 			nullableBytesLen -= savings
 			batchLen -= savings
-			attrs |= int16(codec)
+			r.attrs |= int16(codec)
 			kbin.AppendInt32(dst[:nullableBytesLenAt], nullableBytesLen)
 			kbin.AppendInt32(dst[:batchLenAt], batchLen)
-			kbin.AppendInt16(dst[:attrsAt], attrs)
+			kbin.AppendInt16(dst[:attrsAt], r.attrs)
 		}
 	}
 
@@ -1586,6 +1594,18 @@ func (r seqRecBatch) appendToAsMessageSet(dst []byte, version uint8, compressor
 		)
 	}
 
+	r.attrs = 0
+
+	// Produce request v0 and v1 uses message set v0, which does not have
+	// timestamps. We set bit 8 in our attrs which corresponds with our own
+	// kgo.RecordAttrs's bit. The attrs field is unused in a sink / recBuf
+	// outside of the appending functions or finishing records; if we use
+	// more bits in our internal RecordAttrs, the below will need to
+	// change.
+	if version == 0 || version == 1 {
+		r.attrs |= 0b1000_0000
+	}
+
 	if compressor != nil {
 		toCompress := dst[nullableBytesLenAt+4:] // skip nullable bytes leading prefix
 		w := sliceWriters.Get().(*sliceWriter)
@@ -1601,6 +1621,8 @@ func (r seqRecBatch) appendToAsMessageSet(dst []byte, version uint8, compressor
 		if compressed != nil &&
 			int(wrappedLength) < len(toCompress) {
 
+			r.attrs |= int16(codec)
+
 			dst = appendMessageTo(
 				dst[:nullableBytesLenAt+4],
 				version,