Merge #147543

147543: changefeedccl: include key, size, and mvcc timestamp in Kafka v2 message too large errors r=aerfrei,asg0451 a=elizaMkraule

Kafka v1 error messages for oversized messages included the message key and size, which made it easier to identify problematic data. Kafka v2 was missing this information.

This change restores that missing context. It modifies the Append and Flush method signatures to pass context, allowing the MVCC timestamp to be attached to each Kafka message. The error message for MESSAGE_TOO_LARGE errors is updated to include:
- The message key
- The combined size of the key and value
- The MVCC timestamp
This helps identify what data could not be
delivered.

Fixes #144994
Epic: CRDB-49646

Release note (general change): Kafka v2 changefeed sinks now include the message key, size, and MVCC timestamp in message too large error logs.

Co-authored-by: Eliza Kraule <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/batching_sink.go

@@ -45,7 +45,7 @@ type SinkClient interface {
 // BatchBuffer is an interface to aggregate KVs into a payload that can be sent
 // to the sink.
 type BatchBuffer interface {
-	Append(key []byte, value []byte, attributes attributes)
+	Append(ctx context.Context, key []byte, value []byte, attributes attributes)
 	ShouldFlush() bool
 
 	// Once all data has been Append'ed, Close can be called to return a finalized
@@ -103,6 +103,7 @@ type flushReq struct {
 type attributes struct {
 	tableName string
 	headers   map[string][]byte
+	mvcc      hlc.Timestamp
 }
 
 type rowEvent struct {
@@ -303,14 +304,15 @@ func hashToInt(h hash.Hash32, buf []byte) int {
 }
 
 // Append adds the contents of a kvEvent to the batch, merging its alloc pool.
-func (sb *sinkBatch) Append(e *rowEvent) {
+func (sb *sinkBatch) Append(ctx context.Context, e *rowEvent) {
 	if sb.isEmpty() {
 		sb.bufferTime = timeutil.Now()
 	}
 
-	sb.buffer.Append(e.key, e.val, attributes{
+	sb.buffer.Append(ctx, e.key, e.val, attributes{
 		tableName: e.topicDescriptor.GetTableName(),
 		headers:   e.headers,
+		mvcc:      e.mvcc,
 	})
 
 	sb.keys.Add(hashToInt(sb.hasher, e.key))
@@ -502,7 +504,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 					topicBatches[topic] = batchBuffer
 				}
 
-				batchBuffer.Append(r)
+				batchBuffer.Append(ctx, r)
 				if s.knobs.OnAppend != nil {
 					s.knobs.OnAppend(r)
 				}

pkg/ccl/changefeedccl/sink_kafka_v2.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/admission"
 	"github.com/cockroachdb/cockroach/pkg/util/cidr"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
@@ -164,6 +165,18 @@ func (k *kafkaSinkClientV2) Flush(ctx context.Context, payload SinkPayload) (ret
 				}
 				return nil
 			} else {
+				if len(msgs) == 1 && errors.Is(err, kerr.MessageTooLarge) {
+					msg := msgs[0]
+					mvccVal := msg.Context.Value(mvccTSKey{})
+					var ts hlc.Timestamp
+					if mvccVal != nil {
+						ts = mvccVal.(hlc.Timestamp)
+					}
+					err = errors.Wrapf(err,
+						"Kafka message too large: key=%s size=%d mvcc=%s",
+						string(msg.Key), len(msg.Key)+len(msg.Value), ts,
+					)
+				}
 				return err
 			}
 		}
@@ -301,7 +314,9 @@ type kafkaBuffer struct {
 	batchCfg sinkBatchConfig
 }
 
-func (b *kafkaBuffer) Append(key []byte, value []byte, attrs attributes) {
+type mvccTSKey struct{}
+
+func (b *kafkaBuffer) Append(ctx context.Context, key []byte, value []byte, attrs attributes) {
 	// HACK: kafka sink v1 encodes nil keys as sarama.ByteEncoder(key) which is != nil, and unit tests rely on this.
 	// So do something equivalent.
 	if key == nil {
@@ -313,7 +328,9 @@ func (b *kafkaBuffer) Append(key []byte, value []byte, attrs attributes) {
 		headers = append(headers, kgo.RecordHeader{Key: k, Value: v})
 	}
 
-	b.messages = append(b.messages, &kgo.Record{Key: key, Value: value, Topic: b.topic, Headers: headers})
+	rctx := context.WithValue(ctx, mvccTSKey{}, attrs.mvcc)
+
+	b.messages = append(b.messages, &kgo.Record{Key: key, Value: value, Topic: b.topic, Headers: headers, Context: rctx})
 	b.byteCount += len(value)
 }
 

pkg/ccl/changefeedccl/sink_kafka_v2_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/mocks"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
@@ -99,7 +100,7 @@ func TestKafkaSinkClientV2_Basic(t *testing.T) {
 	buf := fx.sink.MakeBatchBuffer("t")
 	keys := []string{"k1", "k2", "k3"}
 	for i, key := range keys {
-		buf.Append([]byte(key), []byte(strconv.Itoa(i)), attributes{})
+		buf.Append(context.Background(), []byte(key), []byte(strconv.Itoa(i)), attributes{})
 	}
 	payload, err := buf.Close()
 	require.NoError(t, err)
@@ -123,7 +124,13 @@ func TestKafkaSinkClientV2_Resize(t *testing.T) {
 
 		buf := fx.sink.MakeBatchBuffer("t")
 		for i := range 100 {
-			buf.Append([]byte("k1"), []byte(strconv.Itoa(i)), attributes{})
+			buf.Append(context.Background(),
+				[]byte("k1"),
+				[]byte(strconv.Itoa(i)),
+				attributes{
+					mvcc: hlc.Timestamp{WallTime: timeutil.Now().UnixNano()},
+				},
+			)
 		}
 		payload, err := buf.Close()
 		require.NoError(t, err)
@@ -157,7 +164,15 @@ func TestKafkaSinkClientV2_Resize(t *testing.T) {
 			fx.kc.EXPECT().ProduceSync(fx.ctx, payloadAnys[:3]...).Times(1).Return(pr),
 			fx.kc.EXPECT().ProduceSync(fx.ctx, payloadAnys[:1]...).Times(1).Return(pr),
 		)
-		require.Error(t, fx.sink.Flush(fx.ctx, payload))
+
+		err := fx.sink.Flush(fx.ctx, payload)
+		require.Error(t, err)
+
+		// Validate that error includes key, size, and mvcc
+		errStr := err.Error()
+		require.Contains(t, errStr, "k1", "error should include key")
+		require.Regexp(t, `size=\d+`, errStr, "error should include size")
+		require.Regexp(t, `mvcc=\d+`, errStr, "error should include mvcc timestamp")
 	})
 
 	t.Run("resize enabled and it gets everything", func(t *testing.T) {
@@ -556,7 +571,7 @@ func TestKafkaSinkClientV2_ErrorsEventually(t *testing.T) {
 	defer fx.close()
 
 	buf := fx.sink.MakeBatchBuffer("t")
-	buf.Append([]byte("k1"), []byte("v1"), attributes{})
+	buf.Append(context.Background(), []byte("k1"), []byte("v1"), attributes{})
 	payload, err := buf.Close()
 	require.NoError(t, err)
 

pkg/ccl/changefeedccl/sink_pubsub_v2.go

@@ -238,7 +238,9 @@ type pubsubBuffer struct {
 var _ BatchBuffer = (*pubsubBuffer)(nil)
 
 // Append implements the BatchBuffer interface
-func (psb *pubsubBuffer) Append(key []byte, value []byte, attributes attributes) {
+func (psb *pubsubBuffer) Append(
+	ctx context.Context, key []byte, value []byte, attributes attributes,
+) {
 	var content []byte
 	switch psb.sc.format {
 	case changefeedbase.OptFormatJSON:

pkg/ccl/changefeedccl/sink_webhook_v2.go

@@ -356,7 +356,7 @@ type webhookCSVBuffer struct {
 var _ BatchBuffer = (*webhookCSVBuffer)(nil)
 
 // Append implements the BatchBuffer interface.
-func (cb *webhookCSVBuffer) Append(key []byte, value []byte, _ attributes) {
+func (cb *webhookCSVBuffer) Append(ctx context.Context, key []byte, value []byte, _ attributes) {
 	cb.bytes = append(cb.bytes, value...)
 	cb.messageCount += 1
 }
@@ -380,7 +380,7 @@ type webhookJSONBuffer struct {
 var _ BatchBuffer = (*webhookJSONBuffer)(nil)
 
 // Append implements the BatchBuffer interface.
-func (jb *webhookJSONBuffer) Append(key []byte, value []byte, _ attributes) {
+func (jb *webhookJSONBuffer) Append(ctx context.Context, key []byte, value []byte, _ attributes) {
 	jb.messages = append(jb.messages, value)
 	jb.numBytes += len(value)
 }