diff --git a/apis/kafka/v1alpha1/openapi_generated.go b/apis/kafka/v1alpha1/openapi_generated.go
index 418428c3ab..bda6a0c4cd 100644
--- a/apis/kafka/v1alpha1/openapi_generated.go
+++ b/apis/kafka/v1alpha1/openapi_generated.go
@@ -26017,6 +26017,19 @@ func schema_apimachinery_apis_kafka_v1alpha1_RestProxySpec(ref common.ReferenceC
 							Ref:         ref("kmodules.xyz/offshoot-api/api/v2.PodTemplateSpec"),
 						},
 					},
+					"enableSchemaRegistry": {
+						SchemaProps: spec.SchemaProps{
+							Description: "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.",
+							Type:        []string{"boolean"},
+							Format:      "",
+						},
+					},
+					"schemaRegistryRef": {
+						SchemaProps: spec.SchemaProps{
+							Description: "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.",
+							Ref:         ref("kmodules.xyz/client-go/api/v1.ObjectReference"),
+						},
+					},
 					"serviceTemplates": {
 						SchemaProps: spec.SchemaProps{
 							Description: "ServiceTemplates is an optional configuration for services used to expose database",
diff --git a/apis/kafka/v1alpha1/restproxy_types.go b/apis/kafka/v1alpha1/restproxy_types.go
index 8941a850fc..aedac981da 100644
--- a/apis/kafka/v1alpha1/restproxy_types.go
+++ b/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"`
diff --git a/apis/kafka/v1alpha1/restproxy_webhook.go b/apis/kafka/v1alpha1/restproxy_webhook.go
index 7568eb19b3..2853d79414 100644
--- a/apis/kafka/v1alpha1/restproxy_webhook.go
+++ b/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,
diff --git a/apis/kafka/v1alpha1/zz_generated.deepcopy.go b/apis/kafka/v1alpha1/zz_generated.deepcopy.go
index 84e26a739c..f259520e7e 100644
--- a/apis/kafka/v1alpha1/zz_generated.deepcopy.go
+++ b/apis/kafka/v1alpha1/zz_generated.deepcopy.go
@@ -406,6 +406,11 @@ func (in *RestProxySpec) DeepCopyInto(out *RestProxySpec) {
 		**out = **in
 	}
 	in.PodTemplate.DeepCopyInto(&out.PodTemplate)
+	if in.SchemaRegistryRef != nil {
+		in, out := &in.SchemaRegistryRef, &out.SchemaRegistryRef
+		*out = new(v1.ObjectReference)
+		**out = **in
+	}
 	if in.ServiceTemplates != nil {
 		in, out := &in.ServiceTemplates, &out.ServiceTemplates
 		*out = make([]kubedbv1.NamedServiceTemplateSpec, len(*in))
diff --git a/crds/kafka.kubedb.com_restproxies.yaml b/crds/kafka.kubedb.com_restproxies.yaml
index eb1c9668f3..ecf629d5ad 100644
--- a/crds/kafka.kubedb.com_restproxies.yaml
+++ b/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: