Add SchemaRegistry with RestProxy API

Signed-off-by: obaydullahmhs <[email protected]>

apis/kafka/v1alpha1/restproxy_types.go

@@ -70,6 +70,18 @@ type RestProxySpec struct {
 	// +optional
 	PodTemplate ofst.PodTemplateSpec `json:"podTemplate,omitempty"`
 
+	// EnableSchemaRegistry indicates whether the REST Proxy should connect to a Schema Registry.
+	// If set to true, the REST Proxy will establish a connection to the Schema Registry before communicating with Kafka.
+	// This is necessary when producing or consuming messages that use Avro or other schema-based formats.
+	// +optional
+	EnableSchemaRegistry bool `json:"enableSchemaRegistry,omitempty"`
+
+	// SchemaRegistryRef provides a reference to the Schema Registry configuration.
+	// If EnableSchemaRegistry and SchemaRegistryRef are both set, the REST Proxy will connect to the external Schema Registry.
+	// Otherwise, the REST Proxy will use the internal Schema Registry.
+	// +optional
+	SchemaRegistryRef *kmapi.ObjectReference `json:"schemaRegistryRef,omitempty"`
+
 	// ServiceTemplates is an optional configuration for services used to expose database
 	// +optional
 	ServiceTemplates []dbapi.NamedServiceTemplateSpec `json:"serviceTemplates,omitempty"`

apis/kafka/v1alpha1/restproxy_webhook.go

@@ -96,6 +96,14 @@ func (k *RestProxy) ValidateCreateOrUpdate() field.ErrorList {
 		return allErr
 	}
 
+	if !k.Spec.EnableSchemaRegistry {
+		if k.Spec.SchemaRegistryRef != nil {
+			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("enableSchemaRegistry"),
+				k.Name,
+				"SchemaRegistryRef can only be set when EnableSchemaRegistry is true"))
+		}
+	}
+
 	if k.Spec.DeletionPolicy == dbapi.DeletionPolicyHalt {
 		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("deletionPolicy"),
 			k.Name,

crds/kafka.kubedb.com_restproxies.yaml

@@ -55,6 +55,8 @@ spec:
                 - WipeOut
                 - DoNotTerminate
                 type: string
+              enableSchemaRegistry:
+                type: boolean
               healthChecker:
                 default:
                   failureThreshold: 3
@@ -3152,6 +3154,15 @@ spec:
               replicas:
                 format: int32
                 type: integer
+              schemaRegistryRef:
+                properties:
+                  name:
+                    type: string
+                  namespace:
+                    type: string
+                required:
+                - name
+                type: object
               serviceTemplates:
                 items:
                   properties: