produce: set MaxTimestamp

Similar to other clients, set the MaxTimestamp computed from the records
in the batch.

Signed-off-by: Tyler Rockwood <[email protected]>

Author: rockwotj

produce_set.go

@@ -164,9 +164,13 @@ func (ps *produceSet) buildRequest() *ProduceRequest {
 				rb := set.recordsToSend.RecordBatch
 				if len(rb.Records) > 0 {
 					rb.LastOffsetDelta = int32(len(rb.Records) - 1)
+					var maxTimestampDelta time.Duration
 					for i, record := range rb.Records {
 						record.OffsetDelta = int64(i)
+						maxTimestampDelta = max(maxTimestampDelta, record.TimestampDelta)
 					}
+					// Also set the MaxTimestamp similar to the
+					rb.MaxTimestamp = rb.FirstTimestamp.Add(maxTimestampDelta)
 				}
 
 				// Set the batch as transactional when a transactionalID is set

produce_set_test.go

@@ -260,6 +260,9 @@ func TestProduceSetV3RequestBuilding(t *testing.T) {
 	if !batch.FirstTimestamp.Equal(now.Truncate(time.Millisecond)) {
 		t.Errorf("Wrong first timestamp: %v", batch.FirstTimestamp)
 	}
+	if !batch.MaxTimestamp.Equal(now.Add(9 * time.Second).Truncate(time.Millisecond)) {
+		t.Errorf("Wrong max timestamp: %v", batch.MaxTimestamp)
+	}
 	for i := 0; i < 10; i++ {
 		rec := batch.Records[i]
 		if rec.TimestampDelta != time.Duration(i)*time.Second {