diff --git a/ingestion/src/metadata/utils/ssl_manager.py b/ingestion/src/metadata/utils/ssl_manager.py
index a68910c65ba..8d50c354cfc 100644
--- a/ingestion/src/metadata/utils/ssl_manager.py
+++ b/ingestion/src/metadata/utils/ssl_manager.py
@@ -224,11 +224,17 @@ class SSLManager:
                 "ssl.key.location": getattr(self, "key_consumer_config", None),
                 "ssl.certificate.location": getattr(self, "cert_consumer_config", None),
             }
-        connection.schemaRegistryConfig["ssl.ca.location"] = self.ca_file_path
-        connection.schemaRegistryConfig["ssl.key.location"] = self.key_file_path
-        connection.schemaRegistryConfig[
-            "ssl.certificate.location"
-        ] = self.cert_file_path
+        if connection.schemaRegistrySSL:
+            connection.schemaRegistryConfig["ssl.ca.location"] = getattr(
+                self, "ca_schema_registry", None
+            )
+
+            connection.schemaRegistryConfig["ssl.key.location"] = getattr(
+                self, "key_schema_registry", None
+            )
+            connection.schemaRegistryConfig["ssl.certificate.location"] = getattr(
+                self, "cert_schema_registry", None
+            )
         return connection
 
     @setup_ssl.register(CassandraConnection)
diff --git a/ingestion/tests/unit/test_ssl_manager.py b/ingestion/tests/unit/test_ssl_manager.py
index 80516cb383c..a48eac5000c 100644
--- a/ingestion/tests/unit/test_ssl_manager.py
+++ b/ingestion/tests/unit/test_ssl_manager.py
@@ -125,6 +125,21 @@ class KafkaSourceSSLTest(TestCase):
             kafka_source_with_ssl.service_connection.schemaRegistrySSL.root.sslCertificate.get_secret_value(),
             "sslCertificateData",
         )
+        self.assertIsNotNone(
+            kafka_source_with_ssl.service_connection.schemaRegistryConfig.get(
+                "ssl.ca.location"
+            ),
+        )
+        self.assertIsNotNone(
+            kafka_source_with_ssl.service_connection.schemaRegistryConfig.get(
+                "ssl.key.location"
+            ),
+        )
+        self.assertIsNotNone(
+            kafka_source_with_ssl.service_connection.schemaRegistryConfig.get(
+                "ssl.certificate.location"
+            ),
+        )
 
 
 class CassandraSourceSSLTest(TestCase):