From 24a850dac50c48547372e589db89c79bb516104d Mon Sep 17 00:00:00 2001
From: Tomasz Sadura <tomasz@redpanda.com>
Date: Wed, 23 Aug 2023 09:57:03 +0200
Subject: [PATCH] backend: add kafkaKeyFieldName to BigQuery sink connector
 guide

---
 backend/pkg/connector/guide/bigquery_sink.go | 2 ++
 backend/pkg/connector/patch/bigquery_sink.go | 8 +++++---
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/backend/pkg/connector/guide/bigquery_sink.go b/backend/pkg/connector/guide/bigquery_sink.go
index 24269345a..6440e1061 100644
--- a/backend/pkg/connector/guide/bigquery_sink.go
+++ b/backend/pkg/connector/guide/bigquery_sink.go
@@ -44,6 +44,7 @@ func NewBigQuerySinkGuide(opts ...Option) Guide {
 						// No Group name and description here
 						ConfigKeys: append([]string{
 							"key.converter",
+							"key.converter.schemas.enable",
 							"value.converter",
 							"value.converter.schemas.enable",
 							"autoCreateTables",
@@ -52,6 +53,7 @@ func NewBigQuerySinkGuide(opts ...Option) Guide {
 							"allowBigQueryRequiredFieldRelaxation",
 							"upsertEnabled",
 							"deleteEnabled",
+							"kafkaKeyFieldName",
 							"timePartitioningType",
 							"bigQueryRetry",
 							"bigQueryRetryWait",
diff --git a/backend/pkg/connector/patch/bigquery_sink.go b/backend/pkg/connector/patch/bigquery_sink.go
index a1bd55237..80ddb6512 100644
--- a/backend/pkg/connector/patch/bigquery_sink.go
+++ b/backend/pkg/connector/patch/bigquery_sink.go
@@ -51,8 +51,6 @@ func (c *ConfigPatchBigQuery) IsMatch(configKey, connectorClass string) bool {
 func (*ConfigPatchBigQuery) PatchDefinition(d model.ConfigDefinition, _ string) model.ConfigDefinition {
 	// Misc patches
 	switch d.Definition.Name {
-	case keyConverter:
-		d.SetImportance(model.ConfigDefinitionImportanceLow)
 	case valueConverter:
 		d.ClearRecommendedValuesWithMetadata().
 			AddRecommendedValueWithMetadata("io.confluent.connect.avro.AvroConverter", "AVRO").
@@ -82,6 +80,8 @@ func (*ConfigPatchBigQuery) PatchDefinition(d model.ConfigDefinition, _ string)
 		d.SetDisplayName("Upsert enabled")
 	case "deleteEnabled":
 		d.SetDisplayName("Delete enabled")
+	case "kafkaKeyFieldName":
+		d.SetDisplayName("Kafka key field name")
 	case "bigQueryRetry":
 		d.SetDisplayName("BigQuery retry attempts")
 	case "bigQueryRetryWait":
@@ -97,7 +97,9 @@ func (*ConfigPatchBigQuery) PatchDefinition(d model.ConfigDefinition, _ string)
 	switch d.Definition.Name {
 	case "keyfile":
 		d.SetImportance(model.ConfigDefinitionImportanceHigh)
-	case "autoCreateTables":
+	case keyConverter,
+		"autoCreateTables",
+		"kafkaKeyFieldName":
 		d.SetImportance(model.ConfigDefinitionImportanceMedium)
 	}