Allow setting schema registry configuration parameters via the RPK node config block

src/go/rpk/pkg/cli/profile/create.go

@@ -259,6 +259,7 @@ func CreateFlow(
 		p = &config.RpkProfile{
 			KafkaAPI: nodeCfg.KafkaAPI,
 			AdminAPI: nodeCfg.AdminAPI,
+			SR:       nodeCfg.SR,
 		}
 	}
 	if err := doSet(p, set); err != nil {

src/go/rpk/pkg/config/config.go

@@ -247,6 +247,7 @@ func (y *RedpandaYaml) setDevMode() {
 	y.Rpk = RpkNodeConfig{
 		KafkaAPI:             y.Rpk.KafkaAPI,
 		AdminAPI:             y.Rpk.AdminAPI,
+		SR:                   y.Rpk.SR,
 		AdditionalStartFlags: y.Rpk.AdditionalStartFlags,
 		SMP:                  DevDefault().Rpk.SMP,
 		Overprovisioned:      true,

src/go/rpk/pkg/config/params.go

@@ -1389,6 +1389,9 @@ func (c *Config) mergeRpkIntoRedpanda(actual bool) {
 	if !reflect.DeepEqual(p.AdminAPI, RpkAdminAPI{}) {
 		dst.AdminAPI = p.AdminAPI
 	}
+	if !reflect.DeepEqual(p.SR, RpkSchemaRegistryAPI{}) {
+		dst.SR = p.SR
+	}
 }
 
 // This function ensures a current profile exists in the Virtual rpk.yaml.
@@ -1438,6 +1441,9 @@ func (c *Config) ensureBrokerAddrs() {
 		if len(dst.Rpk.AdminAPI.Addresses) == 0 {
 			dst.Rpk.AdminAPI.Addresses = []string{net.JoinHostPort("127.0.0.1", strconv.Itoa(DefaultAdminPort))}
 		}
+		if len(dst.Rpk.SR.Addresses) == 0 {
+			dst.Rpk.SR.Addresses = []string{net.JoinHostPort("127.0.0.1", strconv.Itoa(DefaultSchemaRegPort))}
+		}
 	}
 	{
 		dst := c.rpkYaml.Profile(c.rpkYaml.CurrentProfile) // must exist by this function
@@ -1447,11 +1453,6 @@ func (c *Config) ensureBrokerAddrs() {
 		if len(dst.AdminAPI.Addresses) == 0 {
 			dst.AdminAPI.Addresses = []string{net.JoinHostPort("127.0.0.1", strconv.Itoa(DefaultAdminPort))}
 		}
-	}
-	{
-		dst := c.rpkYaml.Profile(c.rpkYaml.CurrentProfile)
-		// Schema Registry is only supported in profiles. Not in old rpk section
-		// of our redpanda.yaml
 		if len(dst.SR.Addresses) == 0 {
 			dst.SR.Addresses = []string{net.JoinHostPort("127.0.0.1", strconv.Itoa(DefaultSchemaRegPort))}
 		}
@@ -1473,6 +1474,9 @@ func (c *Config) mergeRedpandaIntoRpk() {
 	if reflect.DeepEqual(p.AdminAPI, RpkAdminAPI{}) {
 		p.AdminAPI = src.AdminAPI
 	}
+	if reflect.DeepEqual(p.SR, RpkSchemaRegistryAPI{}) {
+		p.SR = src.SR
+	}
 }
 
 // Similar to backcompat flags, we first capture old env vars and then new

src/go/rpk/pkg/config/params_test.go

@@ -41,6 +41,9 @@ rpk:
     admin_api:
         addresses:
             - 127.0.0.1:9644
+    schema_registry:
+        addresses:
+            - 127.0.0.1:8081
 `,
 		},
 		{
@@ -100,6 +103,9 @@ rpk:
             - 127.0.0.1:4444
             - 10.0.0.1:4444
             - 122.65.33.12:4444
+    schema_registry:
+        addresses:
+            - 127.0.0.1:8081
 `,
 		},
 		{
@@ -117,6 +123,9 @@ rpk:
     kafka_api:
          brokers:
              - 127.0.0.1:9092
+    schema_registry:
+         addresses:
+             - 127.0.0.1:8081
 `,
 			exp: `redpanda:
     seed_servers: []
@@ -131,6 +140,9 @@ rpk:
     kafka_api:
          brokers:
              - 127.0.0.1:9092
+    schema_registry:
+         addresses:
+             - 127.0.0.1:8081
 `,
 		},
 		{
@@ -151,6 +163,9 @@ rpk:
     kafka_api:
          brokers:
              - 127.0.0.1:9092
+    schema_registry:
+         addresses:
+             - 127.0.0.1:8081
 `,
 			exp: `redpanda:
     data_directory: /var/lib/redpanda/data
@@ -168,6 +183,9 @@ rpk:
     admin_api:
         addresses:
             - 127.0.0.1:9644
+    schema_registry:
+        addresses:
+            - 127.0.0.1:8081
 `,
 		},
 	}
@@ -718,6 +736,9 @@ rpk:
     admin_api:
         addresses:
             - 127.0.0.1:9644
+    schema_registry:
+        addresses:
+            - 127.0.0.1:8081
     overprovisioned: true
     coredump_dir: /var/lib/redpanda/coredump
 pandaproxy: {}
@@ -781,6 +802,9 @@ cloud_auth:
         - address: 0.0.0.3
           port: 9092
 rpk:
+    schema_registry:
+        addresses:
+            - 127.0.0.1:3232
     enable_memory_locking: true
     tune_network: true
     tune_disk_scheduler: true
@@ -808,6 +832,9 @@ rpk:
     admin_api:
         addresses:
             - 0.0.0.3:9644
+    schema_registry:
+        addresses:
+            - 127.0.0.1:3232
     enable_memory_locking: true
     tune_network: true
     tune_disk_scheduler: true
@@ -841,7 +868,7 @@ profiles:
             - 0.0.0.3:9644
       schema_registry:
         addresses:
-            - 127.0.0.1:8081
+            - 127.0.0.1:3232
 cloud_auth:
     - name: default
       organization: Default organization
@@ -909,6 +936,9 @@ rpk:
     admin_api:
         addresses:
             - 0.0.0.3:9644
+    schema_registry:
+        addresses:
+            - 0.0.0.2:8081
     overprovisioned: true
     coredump_dir: /var/lib/redpanda/coredump
 pandaproxy: {}
@@ -968,6 +998,9 @@ cloud_auth:
         - address: admin.com
           port: 4444
 rpk:
+    schema_registry:
+        addresses:
+            - 127.0.0.1:8081
     enable_memory_locking: true
     tune_network: true
     tune_disk_scheduler: true
@@ -997,7 +1030,9 @@ profiles:
         brokers:
             - 128.0.0.4
       admin_api: {}
-      schema_registry: {}
+      schema_registry:
+        addresses:
+            - 127.0.0.1:3232
 cloud_auth: []
 `,
 
@@ -1017,6 +1052,9 @@ rpk:
     admin_api:
         addresses:
             - 128.0.0.4:9644
+    schema_registry:
+        addresses:
+            - 127.0.0.1:3232
     enable_memory_locking: true
     tune_network: true
     tune_disk_scheduler: true
@@ -1051,7 +1089,7 @@ profiles:
             - 128.0.0.4:9644
       schema_registry:
         addresses:
-            - 127.0.0.1:8081
+            - 127.0.0.1:3232
 cloud_auth:
     - name: default
       organization: Default organization

src/go/rpk/pkg/config/redpanda_yaml.go

@@ -141,8 +141,9 @@ type (
 	}
 
 	RpkNodeConfig struct {
-		KafkaAPI RpkKafkaAPI `yaml:"kafka_api,omitempty" json:"kafka_api"`
-		AdminAPI RpkAdminAPI `yaml:"admin_api,omitempty" json:"admin_api"`
+		KafkaAPI RpkKafkaAPI          `yaml:"kafka_api,omitempty" json:"kafka_api"`
+		AdminAPI RpkAdminAPI          `yaml:"admin_api,omitempty" json:"admin_api"`
+		SR       RpkSchemaRegistryAPI `yaml:"schema_registry,omitempty" json:"schema_registry"`
 
 		// The following four configs are passed to redpanda on `rpk
 		// redpanda start`. They are not tuner configs. They live here

src/go/rpk/pkg/config/weak.go

@@ -411,29 +411,30 @@ func (rpkc *RpkNodeConfig) UnmarshalYAML(n *yaml.Node) error {
 		// Deprecated 2021-07-1
 		SASL *SASL `yaml:"sasl"`
 
-		KafkaAPI                 RpkKafkaAPI     `yaml:"kafka_api"`
-		AdminAPI                 RpkAdminAPI     `yaml:"admin_api"`
-		AdditionalStartFlags     weakStringArray `yaml:"additional_start_flags"`
-		TuneNetwork              weakBool        `yaml:"tune_network"`
-		TuneDiskScheduler        weakBool        `yaml:"tune_disk_scheduler"`
-		TuneNomerges             weakBool        `yaml:"tune_disk_nomerges"`
-		TuneDiskWriteCache       weakBool        `yaml:"tune_disk_write_cache"`
-		TuneDiskIrq              weakBool        `yaml:"tune_disk_irq"`
-		TuneFstrim               weakBool        `yaml:"tune_fstrim"`
-		TuneCPU                  weakBool        `yaml:"tune_cpu"`
-		TuneAioEvents            weakBool        `yaml:"tune_aio_events"`
-		TuneClocksource          weakBool        `yaml:"tune_clocksource"`
-		TuneSwappiness           weakBool        `yaml:"tune_swappiness"`
-		TuneTransparentHugePages weakBool        `yaml:"tune_transparent_hugepages"`
-		EnableMemoryLocking      weakBool        `yaml:"enable_memory_locking"`
-		TuneCoredump             weakBool        `yaml:"tune_coredump"`
-		CoredumpDir              weakString      `yaml:"coredump_dir"`
-		TuneBallastFile          weakBool        `yaml:"tune_ballast_file"`
-		BallastFilePath          weakString      `yaml:"ballast_file_path"`
-		BallastFileSize          weakString      `yaml:"ballast_file_size"`
-		WellKnownIo              weakString      `yaml:"well_known_io"`
-		Overprovisioned          weakBool        `yaml:"overprovisioned"`
-		SMP                      *weakInt        `yaml:"smp"`
+		KafkaAPI                 RpkKafkaAPI          `yaml:"kafka_api"`
+		AdminAPI                 RpkAdminAPI          `yaml:"admin_api"`
+		SR                       RpkSchemaRegistryAPI `yaml:"schema_registry"`
+		AdditionalStartFlags     weakStringArray      `yaml:"additional_start_flags"`
+		TuneNetwork              weakBool             `yaml:"tune_network"`
+		TuneDiskScheduler        weakBool             `yaml:"tune_disk_scheduler"`
+		TuneNomerges             weakBool             `yaml:"tune_disk_nomerges"`
+		TuneDiskWriteCache       weakBool             `yaml:"tune_disk_write_cache"`
+		TuneDiskIrq              weakBool             `yaml:"tune_disk_irq"`
+		TuneFstrim               weakBool             `yaml:"tune_fstrim"`
+		TuneCPU                  weakBool             `yaml:"tune_cpu"`
+		TuneAioEvents            weakBool             `yaml:"tune_aio_events"`
+		TuneClocksource          weakBool             `yaml:"tune_clocksource"`
+		TuneSwappiness           weakBool             `yaml:"tune_swappiness"`
+		TuneTransparentHugePages weakBool             `yaml:"tune_transparent_hugepages"`
+		EnableMemoryLocking      weakBool             `yaml:"enable_memory_locking"`
+		TuneCoredump             weakBool             `yaml:"tune_coredump"`
+		CoredumpDir              weakString           `yaml:"coredump_dir"`
+		TuneBallastFile          weakBool             `yaml:"tune_ballast_file"`
+		BallastFilePath          weakString           `yaml:"ballast_file_path"`
+		BallastFileSize          weakString           `yaml:"ballast_file_size"`
+		WellKnownIo              weakString           `yaml:"well_known_io"`
+		Overprovisioned          weakBool             `yaml:"overprovisioned"`
+		SMP                      *weakInt             `yaml:"smp"`
 	}
 	if err := n.Decode(&internal); err != nil {
 		return err
@@ -442,6 +443,7 @@ func (rpkc *RpkNodeConfig) UnmarshalYAML(n *yaml.Node) error {
 	// backcompat, immediately convert to new tls
 	rpkc.KafkaAPI = internal.KafkaAPI
 	rpkc.AdminAPI = internal.AdminAPI
+	rpkc.SR = internal.SR
 	if rpkc.KafkaAPI.TLS == nil {
 		rpkc.KafkaAPI.TLS = internal.TLS
 	}
@@ -451,6 +453,9 @@ func (rpkc *RpkNodeConfig) UnmarshalYAML(n *yaml.Node) error {
 	if rpkc.AdminAPI.TLS == nil {
 		rpkc.AdminAPI.TLS = internal.TLS
 	}
+	if rpkc.SR.TLS == nil {
+		rpkc.SR.TLS = internal.TLS
+	}
 	rpkc.AdditionalStartFlags = internal.AdditionalStartFlags
 	rpkc.EnableMemoryLocking = bool(internal.EnableMemoryLocking)
 	rpkc.Overprovisioned = bool(internal.Overprovisioned)
@@ -503,6 +508,19 @@ func (r *RpkAdminAPI) UnmarshalYAML(n *yaml.Node) error {
 	return nil
 }
 
+func (r *RpkSchemaRegistryAPI) UnmarshalYAML(n *yaml.Node) error {
+	var internal struct {
+		Addresses weakStringArray `yaml:"addresses"`
+		TLS       *TLS            `yaml:"tls"`
+	}
+	if err := n.Decode(&internal); err != nil {
+		return err
+	}
+	r.Addresses = internal.Addresses
+	r.TLS = internal.TLS
+	return nil
+}
+
 func (p *Pandaproxy) UnmarshalYAML(n *yaml.Node) error {
 	var internal struct {
 		PandaproxyAPI           namedAuthNSocketAddresses `yaml:"pandaproxy_api"`

src/go/rpk/pkg/config/weak_test.go

@@ -973,6 +973,13 @@ rpk:
     tls:
       cert_file: ~/certs/admin-cert.pem
       truststore_file: ~/certs/admin-ca.pem
+  schema_registry:
+    addresses:
+    - 192.168.72.34:8081
+    - 192.168.72.35:8081
+    tls:
+      cert_file: ~/certs/admin-cert.pem
+      truststore_file: ~/certs/admin-ca.pem
   tune_network: false
   tune_disk_scheduler: false
   tune_cpu: true
@@ -1071,6 +1078,10 @@ rpk:
 						Addresses: []string{"192.168.72.34:9644", "192.168.72.35:9644"},
 						TLS:       &TLS{CertFile: "~/certs/admin-cert.pem", TruststoreFile: "~/certs/admin-ca.pem"},
 					},
+					SR: RpkSchemaRegistryAPI{
+						Addresses: []string{"192.168.72.34:8081", "192.168.72.35:8081"},
+						TLS:       &TLS{CertFile: "~/certs/admin-cert.pem", TruststoreFile: "~/certs/admin-ca.pem"},
+					},
 					Tuners: RpkNodeTuners{
 						TuneNetwork:       false,
 						TuneDiskScheduler: false,
@@ -1293,6 +1304,13 @@ rpk:
     tls:
       cert_file: ~/certs/admin-cert.pem
       truststore_file: ~/certs/admin-ca.pem
+  schema_registry:
+    addresses:
+    - 192.168.72.34:8081
+    - 192.168.72.35:8081
+    tls:
+      cert_file: ~/certs/admin-cert.pem
+      truststore_file: ~/certs/admin-ca.pem
   tune_network: false
   tune_disk_scheduler: false
   tune_cpu: 1
@@ -1390,6 +1408,10 @@ rpk:
 						Addresses: []string{"192.168.72.34:9644", "192.168.72.35:9644"},
 						TLS:       &TLS{CertFile: "~/certs/admin-cert.pem", TruststoreFile: "~/certs/admin-ca.pem"},
 					},
+					SR: RpkSchemaRegistryAPI{
+						Addresses: []string{"192.168.72.34:8081", "192.168.72.35:8081"},
+						TLS:       &TLS{CertFile: "~/certs/admin-cert.pem", TruststoreFile: "~/certs/admin-ca.pem"},
+					},
 					Tuners: RpkNodeTuners{
 						TuneNetwork:       false,
 						TuneDiskScheduler: false,