changefeedccl: include key, size, and mvcc timestamp in Kafka v2

message too large errors

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 also updated to
include:
- The message key
- The combined size of the key and value
- The MVCC timestamp
This helps users 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.

Author: elizaMkraule

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