changefeedccl: support for constant headers for webhook and kafka sinks

This adds support for constant headers for webhook
and kafka sinks, to be specified in the CREATE
CHANGEFEED statement as a JSON object.

Part of: #142273
Epic: CRDB-48880

Release note (general change): The CREATE
CHANGEFEED statement now supports the
`extra_headers` option, which can be used to
specify extra headers for webhook and kafka
sinks. This can be used to add headers to all
messages sent to the sink.

Author: asg0451

pkg/ccl/changefeedccl/cdctest/mock_webhook_sink.go

@@ -17,6 +17,11 @@ import (
 	"github.com/klauspost/compress/zstd"
 )
 
+type RowWithHeaders struct {
+	Row     string
+	Headers http.Header
+}
+
 // MockWebhookSink is the Webhook sink used in tests.
 type MockWebhookSink struct {
 	basicAuth          bool
@@ -28,7 +33,7 @@ type MockWebhookSink struct {
 		responseBodies   map[int][]byte
 		statusCodes      []int
 		statusCodesIndex int
-		rows             []string
+		rows             []RowWithHeaders
 		lastHeaders      http.Header
 		notify           chan struct{}
 	}
@@ -160,7 +165,7 @@ func (s *MockWebhookSink) Latest() string {
 		return ""
 	}
 	latest := s.mu.rows[len(s.mu.rows)-1]
-	return latest
+	return latest.Row
 }
 
 // Pop deletes and returns the oldest message from MockWebhookSink.
@@ -170,11 +175,23 @@ func (s *MockWebhookSink) Pop() string {
 	if len(s.mu.rows) > 0 {
 		oldest := s.mu.rows[0]
 		s.mu.rows = s.mu.rows[1:]
-		return oldest
+		return oldest.Row
 	}
 	return ""
 }
 
+// PopWithHeaders deletes and returns the oldest message from MockWebhookSink along with its headers.
+func (s *MockWebhookSink) PopWithHeaders() RowWithHeaders {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.mu.rows) > 0 {
+		oldest := s.mu.rows[0]
+		s.mu.rows = s.mu.rows[1:]
+		return oldest
+	}
+	return RowWithHeaders{}
+}
+
 // NotifyMessage arranges for channel to be closed when message arrives.
 func (s *MockWebhookSink) NotifyMessage() chan struct{} {
 	c := make(chan struct{})
@@ -298,7 +315,7 @@ func (s *MockWebhookSink) publish(hw http.ResponseWriter, hr *http.Request) erro
 	}
 
 	s.mu.Lock()
-	s.mu.rows = append(s.mu.rows, string(row))
+	s.mu.rows = append(s.mu.rows, RowWithHeaders{Row: string(row), Headers: hr.Header})
 	if s.mu.notify != nil {
 		close(s.mu.notify)
 		s.mu.notify = nil

pkg/ccl/changefeedccl/cdctest/testfeed.go

@@ -7,6 +7,8 @@ package cdctest
 
 import (
 	"fmt"
+	"slices"
+	"strings"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -39,7 +41,10 @@ func (h Headers) String() string {
 		return ""
 	}
 	s := "("
-	for _, v := range h {
+	sorted := slices.SortedFunc(slices.Values(h), func(a, b Header) int {
+		return strings.Compare(a.K, b.K)
+	})
+	for _, v := range sorted {
 		s += fmt.Sprintf("%s: %s, ", v.K, v.V)
 	}
 	return s[:len(s)-2] + ")"

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11821,6 +11821,91 @@ func TestCloudstorageParallelCompression(t *testing.T) {
 	})
 }
 
+func TestChangefeedExtraHeaders(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		// Headers are not supported in the v1 kafka sink.
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.new_kafka_sink.enabled = true`)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (key INT PRIMARY KEY);`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1);`)
+
+		cases := []struct {
+			name        string
+			headersArg  string
+			wantHeaders cdctest.Headers
+			expectErr   bool
+		}{
+			{
+				name:        "single header",
+				headersArg:  `{"X-Someheader": "somevalue"}`,
+				wantHeaders: cdctest.Headers{{K: "X-Someheader", V: []byte("somevalue")}},
+			},
+			{
+				name:       "multiple headers",
+				headersArg: `{"X-Someheader": "somevalue", "X-Someotherheader": "someothervalue"}`,
+				wantHeaders: cdctest.Headers{
+					{K: "X-Someheader", V: []byte("somevalue")},
+					{K: "X-Someotherheader", V: []byte("someothervalue")},
+				},
+			},
+			{
+				name:       "inappropriate json",
+				headersArg: `4`,
+				expectErr:  true,
+			},
+			{
+				name:       "also inappropriate json",
+				headersArg: `["X-Someheader", "somevalue"]`,
+				expectErr:  true,
+			},
+			{
+				name:       "invalid json",
+				headersArg: `xxxx`,
+				expectErr:  true,
+			},
+		}
+
+		for _, c := range cases {
+			feed, err := f.Feed(fmt.Sprintf(`CREATE CHANGEFEED FOR foo WITH extra_headers='%s'`, c.headersArg))
+			if c.expectErr {
+				require.Error(t, err)
+				continue
+			} else {
+				require.NoError(t, err)
+			}
+
+			assertPayloads(t, feed, []string{
+				fmt.Sprintf(`foo: [1]%s->{"after": {"key": 1}}`, c.wantHeaders.String()),
+			})
+			closeFeed(t, feed)
+		}
+	}
+
+	cdcTest(t, testFn, feedTestRestrictSinks("kafka", "webhook"))
+}
+func TestChangefeedAdditionalHeadersDoesntWorkWithV1KafkaSink(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.new_kafka_sink.enabled = false`)
+
+		sqlDB.Exec(t, `CREATE TABLE foo (key INT PRIMARY KEY);`)
+		sqlDB.Exec(t, `INSERT INTO foo VALUES (1);`)
+
+		_, err := f.Feed(`CREATE CHANGEFEED FOR foo WITH extra_headers='{"X-Someheader": "somevalue"}'`)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "headers are not supported for the v1 kafka sink")
+	}
+
+	cdcTest(t, testFn, feedTestForceSink("kafka"))
+}
+
 func TestDatabaseLevelChangefeed(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

pkg/ccl/changefeedccl/changefeedbase/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//pkg/sql/types",
         "//pkg/util",
         "//pkg/util/iterutil",
+        "//pkg/util/json",
         "//pkg/util/metamorphic",
         "@com_github_cockroachdb_errors//:errors",
     ],

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/util/json"
 	"github.com/cockroachdb/errors"
 )
 
@@ -117,6 +118,7 @@ const (
 	// TODO(#142273): look into whether we want to add headers to pub/sub, and other
 	// sinks as well (eg cloudstorage, webhook, ..). Currently it's kafka-only.
 	OptHeadersJSONColumnName = `headers_json_column_name`
+	OptExtraHeaders          = `extra_headers`
 
 	OptVirtualColumnsOmitted VirtualColumnVisibility = `omitted`
 	OptVirtualColumnsNull    VirtualColumnVisibility = `null`
@@ -408,6 +410,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptEncodeJSONValueNullAsObject:        flagOption,
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
 	OptHeadersJSONColumnName:              stringOption,
+	OptExtraHeaders:                       jsonOption,
 }
 
 // CommonOptions is options common to all sinks
@@ -428,13 +431,13 @@ var CommonOptions = makeStringSet(OptCursor, OptEndTime, OptEnvelope,
 var SQLValidOptions map[string]struct{} = nil
 
 // KafkaValidOptions is options exclusive to Kafka sink
-var KafkaValidOptions = makeStringSet(OptAvroSchemaPrefix, OptConfluentSchemaRegistry, OptKafkaSinkConfig, OptHeadersJSONColumnName)
+var KafkaValidOptions = makeStringSet(OptAvroSchemaPrefix, OptConfluentSchemaRegistry, OptKafkaSinkConfig, OptHeadersJSONColumnName, OptExtraHeaders)
 
 // CloudStorageValidOptions is options exclusive to cloud storage sink
 var CloudStorageValidOptions = makeStringSet(OptCompression)
 
 // WebhookValidOptions is options exclusive to webhook sink
-var WebhookValidOptions = makeStringSet(OptWebhookAuthHeader, OptWebhookClientTimeout, OptWebhookSinkConfig, OptCompression)
+var WebhookValidOptions = makeStringSet(OptWebhookAuthHeader, OptWebhookClientTimeout, OptWebhookSinkConfig, OptCompression, OptExtraHeaders)
 
 // PubsubValidOptions is options exclusive to pubsub sink
 var PubsubValidOptions = makeStringSet(OptPubsubSinkConfig)
@@ -477,6 +480,7 @@ func RedactUserFromURI(uri string) (string, error) {
 // RedactedOptions are options whose values should be replaced with "redacted" in job descriptions and errors.
 var RedactedOptions = map[string]redactionFunc{
 	OptWebhookAuthHeader:       redactSimple,
+	OptExtraHeaders:            redactSimple,
 	SinkParamClientKey:         redactSimple,
 	OptConfluentSchemaRegistry: RedactUserFromURI,
 }
@@ -1032,6 +1036,7 @@ func (s StatementOptions) GetFilters() Filters {
 type WebhookSinkOptions struct {
 	JSONConfig    SinkSpecificJSONConfig
 	AuthHeader    string
+	ExtraHeaders  map[string]string
 	ClientTimeout *time.Duration
 	Compression   string
 }
@@ -1049,13 +1054,66 @@ func (s StatementOptions) GetWebhookSinkOptions() (WebhookSinkOptions, error) {
 		return o, err
 	}
 	o.ClientTimeout = timeout
+
+	headersMap, err := parseHeaders[string](s.m[OptExtraHeaders])
+	if err != nil {
+		return o, err
+	}
+	o.ExtraHeaders = headersMap
 	return o, nil
 }
 
-// GetKafkaConfigJSON returns arbitrary json to be interpreted
-// by the kafka sink.
-func (s StatementOptions) GetKafkaConfigJSON() SinkSpecificJSONConfig {
-	return s.getJSONValue(OptKafkaSinkConfig)
+func parseHeaders[S interface{ string | []byte }](headers string) (map[string]S, error) {
+	if headers == "" {
+		return nil, nil
+	}
+	headersJ, err := json.ParseJSON(headers)
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing headers")
+	}
+	it, err := headersJ.ObjectIter()
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing headers as object")
+	}
+	if it == nil {
+		return nil, errors.Newf("headers is not a JSON object: %s", headers)
+	}
+	headersMap := make(map[string]S, headersJ.Len())
+	for it.Next() {
+		k := it.Key()
+		v := it.Value()
+		s, err := v.AsText()
+		if err != nil {
+			return nil, errors.Wrap(err, "parsing header value as text")
+		}
+		if s == nil {
+			continue
+		}
+		headersMap[k] = S(*s)
+	}
+	return headersMap, nil
+}
+
+type KafkaSinkOptions struct {
+	// JSONConfig is arbitrary json to be interpreted
+	// by the kafka sink.
+	JSONConfig SinkSpecificJSONConfig
+
+	// Headers is a map of header names to values.
+	Headers map[string][]byte
+}
+
+func (s StatementOptions) GetKafkaSinkOptions() (KafkaSinkOptions, error) {
+	headersMap, err := parseHeaders[[]byte](s.m[OptExtraHeaders])
+	if err != nil {
+		return KafkaSinkOptions{}, err
+	}
+
+	o := KafkaSinkOptions{
+		JSONConfig: s.getJSONValue(OptKafkaSinkConfig),
+		Headers:    headersMap,
+	}
+	return o, nil
 }
 
 // GetPubsubConfigJSON returns arbitrary json to be interpreted

pkg/ccl/changefeedccl/sink.go

@@ -249,20 +249,28 @@ func getSink(
 			return makeNullSink(&changefeedbase.SinkURL{URL: u}, metricsBuilder(nullIsAccounted))
 		case isKafkaSink(u):
 			return validateOptionsAndMakeSink(changefeedbase.KafkaValidOptions, func() (Sink, error) {
+				sinkOpts, err := opts.GetKafkaSinkOptions()
+				if err != nil {
+					return nil, err
+				}
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
-					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), sinkOpts,
 						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
 						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
 				} else {
-					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(), serverCfg.Settings, metricsBuilder)
+					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), sinkOpts, serverCfg.Settings, metricsBuilder)
 				}
 			})
 		case isPulsarSink(u):
 			var testingKnobs *TestingKnobs
 			if knobs, ok := serverCfg.TestingKnobs.Changefeed.(*TestingKnobs); ok {
 				testingKnobs = knobs
 			}
-			return makePulsarSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
+			sinkOpts, err := opts.GetKafkaSinkOptions()
+			if err != nil {
+				return nil, err
+			}
+			return makePulsarSink(ctx, &changefeedbase.SinkURL{URL: u}, encodingOpts, AllTargets(feedCfg), sinkOpts.JSONConfig,
 				serverCfg.Settings, metricsBuilder, testingKnobs)
 		case isWebhookSink(u):
 			webhookOpts, err := opts.GetWebhookSinkOptions()

pkg/ccl/changefeedccl/sink_kafka.go

@@ -1200,7 +1200,7 @@ func makeKafkaSink(
 	ctx context.Context,
 	u *changefeedbase.SinkURL,
 	targets changefeedbase.Targets,
-	jsonStr changefeedbase.SinkSpecificJSONConfig,
+	sinkOpts changefeedbase.KafkaSinkOptions,
 	settings *cluster.Settings,
 	mb metricsRecorderBuilder,
 ) (Sink, error) {
@@ -1210,6 +1210,12 @@ func makeKafkaSink(
 		return nil, errors.Errorf(`%s is not yet supported`, changefeedbase.SinkParamSchemaTopic)
 	}
 
+	jsonStr := sinkOpts.JSONConfig
+	if len(sinkOpts.Headers) > 0 {
+		return nil, errors.Newf("headers are not supported for the v1 kafka sink;"+
+			" use the v2 sink instead via the `%s` cluster setting", KafkaV2Enabled.Name())
+	}
+
 	m := mb(requiresResourceAccounting)
 	config, err := buildKafkaConfig(ctx, u, jsonStr, m.getKafkaThrottlingMetrics(settings), m.netMetrics())
 	if err != nil {

pkg/ccl/changefeedccl/sink_kafka_connection_test.go

@@ -403,7 +403,7 @@ func TestAzureKafkaDefaults(t *testing.T) {
 
 	assertExpectedKgoOpts := func(exp expectation, opts []kgo.Opt) {
 		sinkClient, err := newKafkaSinkClientV2(ctx, opts, sinkBatchConfig{},
-			"", cluster.MakeTestingClusterSettings(), kafkaSinkV2Knobs{}, nilMetricsRecorderBuilder, nil)
+			"", cluster.MakeTestingClusterSettings(), kafkaSinkV2Knobs{}, nilMetricsRecorderBuilder, nil, nil)
 		require.NoError(t, err)
 		defer func() { require.NoError(t, sinkClient.Close()) }()
 		client := sinkClient.client.(*kgo.Client)