option-and-thinking

Author: asg0451

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -114,6 +114,7 @@ const (
 	OptLaggingRangesPollingInterval       = `lagging_ranges_polling_interval`
 	OptIgnoreDisableChangefeedReplication = `ignore_disable_changefeed_replication`
 	OptEncodeJSONValueNullAsObject        = `encode_json_value_null_as_object`
+	OptCreateKafkaTopics                  = `create_kafka_topics`
 	// 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`
@@ -406,6 +407,7 @@ var ChangefeedOptionExpectValues = map[string]OptionPermittedValues{
 	OptLaggingRangesPollingInterval:       durationOption,
 	OptIgnoreDisableChangefeedReplication: flagOption,
 	OptEncodeJSONValueNullAsObject:        flagOption,
+	OptCreateKafkaTopics:                  enum("yes", "no", "auto").orEmptyMeans("auto"),
 	OptEnrichedProperties:                 csv(string(EnrichedPropertySource), string(EnrichedPropertySchema)),
 	OptHeadersJSONColumnName:              stringOption,
 }
@@ -428,7 +430,7 @@ 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, OptCreateKafkaTopics)
 
 // CloudStorageValidOptions is options exclusive to cloud storage sink
 var CloudStorageValidOptions = makeStringSet(OptCompression)
@@ -660,6 +662,26 @@ func (s StatementOptions) GetEndTime() string {
 	return s.m[OptEndTime]
 }
 
+type CreateKafkaTopics string
+
+const (
+	CreateKafkaTopicsAuto CreateKafkaTopics = "auto"
+	CreateKafkaTopicsYes  CreateKafkaTopics = "yes"
+	CreateKafkaTopicsNo   CreateKafkaTopics = "no"
+)
+
+func (s StatementOptions) GetCreateKafkaTopics() (CreateKafkaTopics, error ) {
+	if _, ok := s.m[OptCreateKafkaTopics]; !ok {
+		return CreateKafkaTopicsAuto, nil
+	}
+
+	rawVal, err := s.getEnumValue(OptCreateKafkaTopics)
+	if err != nil {
+		return CreateKafkaTopicsAuto, err
+	}
+	return CreateKafkaTopics(rawVal), nil
+}
+
 func (s StatementOptions) getEnumValue(k string) (string, error) {
 	enumOptions := ChangefeedOptionExpectValues[k]
 	rawVal, present := s.m[k]

pkg/ccl/changefeedccl/sink.go

@@ -250,9 +250,13 @@ func getSink(
 		case isKafkaSink(u):
 			return validateOptionsAndMakeSink(changefeedbase.KafkaValidOptions, func() (Sink, error) {
 				if KafkaV2Enabled.Get(&serverCfg.Settings.SV) {
+					createTopics, err := opts.GetCreateKafkaTopics()
+					if err != nil {
+						return nil, err
+					}
 					return makeKafkaSinkV2(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(),
 						numSinkIOWorkers(serverCfg), newCPUPacerFactory(ctx, serverCfg), timeutil.DefaultTimeSource{},
-						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{})
+						serverCfg.Settings, metricsBuilder, kafkaSinkV2Knobs{}, createTopics)
 				} else {
 					return makeKafkaSink(ctx, &changefeedbase.SinkURL{URL: u}, AllTargets(feedCfg), opts.GetKafkaConfigJSON(), serverCfg.Settings, metricsBuilder)
 				}

pkg/ccl/changefeedccl/sink_kafka_v2.go

@@ -47,6 +47,7 @@ type kafkaSinkClientV2 struct {
 	recordResize        func(numRecords int64)
 
 	topicsForConnectionCheck []string
+	createTopicsManually     bool
 
 	// we need to fetch and keep track of this ourselves since kgo doesnt expose metadata to us
 	metadataMu struct {
@@ -68,6 +69,7 @@ func newKafkaSinkClientV2(
 	knobs kafkaSinkV2Knobs,
 	mb metricsRecorderBuilder,
 	topicsForConnectionCheck []string,
+	createTopics changefeedbase.CreateKafkaTopics,
 ) (*kafkaSinkClientV2, error) {
 	bootstrapBrokers := strings.Split(bootstrapAddrsStr, `,`)
 
@@ -82,8 +84,6 @@ func newKafkaSinkClientV2(
 		kgo.ProducerBatchMaxBytes(256 << 20), // 256MiB
 		kgo.BrokerMaxWriteBytes(1 << 30),     // 1GiB
 
-		kgo.AllowAutoTopicCreation(),
-
 		kgo.RecordRetries(5),
 		// This applies only to non-produce requests, ie the ListTopics call.
 		kgo.RequestRetries(5),
@@ -95,6 +95,10 @@ func newKafkaSinkClientV2(
 		}),
 	}
 
+	if createTopics == changefeedbase.CreateKafkaTopicsAuto {
+		baseOpts = append(baseOpts, kgo.AllowAutoTopicCreation())
+	}
+
 	recordResize := func(numRecords int64) {}
 	if m := mb(requiresResourceAccounting); m != nil { // `m` can be nil in tests.
 		baseOpts = append(baseOpts, kgo.WithHooks(&kgoMetricsAdapter{throttling: m.getKafkaThrottlingMetrics(settings)}))
@@ -128,6 +132,7 @@ func newKafkaSinkClientV2(
 		includeErrorDetails:      changefeedbase.KafkaV2ErrorDetailsEnabled.Get(&settings.SV),
 		recordResize:             recordResize,
 		topicsForConnectionCheck: topicsForConnectionCheck,
+		createTopicsManually:     createTopics == changefeedbase.CreateKafkaTopicsYes,
 	}
 	c.metadataMu.allTopicPartitions = make(map[string][]int32)
 
@@ -361,6 +366,7 @@ func makeKafkaSinkV2(
 	settings *cluster.Settings,
 	mb metricsRecorderBuilder,
 	knobs kafkaSinkV2Knobs,
+	createTopics changefeedbase.CreateKafkaTopics,
 ) (Sink, error) {
 	batchCfg, retryOpts, err := getSinkConfigFromJson(jsonConfig, sinkJSONConfig{
 		// Defaults from the v1 sink - flush immediately.
@@ -400,7 +406,7 @@ func makeKafkaSinkV2(
 	}
 
 	topicsForConnectionCheck := topicNamer.DisplayNamesSlice()
-	client, err := newKafkaSinkClientV2(ctx, clientOpts, batchCfg, u.Host, settings, knobs, mb, topicsForConnectionCheck)
+	client, err := newKafkaSinkClientV2(ctx, clientOpts, batchCfg, u.Host, settings, knobs, mb, topicsForConnectionCheck, createTopics)
 	if err != nil {
 		return nil, err
 	}