Fix validation of SSL CA Certificate for DEK Registry client

Author: rayokota

src/Confluent.SchemaRegistry.Encryption/CachedDekRegistryClient.cs

@@ -236,8 +236,17 @@ public CachedDekRegistryClient(IEnumerable<KeyValuePair<string, string>> config,
                     $"Configured value for {SchemaRegistryConfig.PropertyNames.EnableSslCertificateVerification} must be a bool.");
             }
 
-            this.restService = new DekRestService(schemaRegistryUris, timeoutMs, authenticationHeaderValueProvider,
-                SetSslConfig(config), sslVerify);
+            var sslCaLocation = config.FirstOrDefault(prop => prop.Key.ToLower() == SchemaRegistryConfig.PropertyNames.SslCaLocation).Value;
+            if (string.IsNullOrEmpty(sslCaLocation))
+            {
+                this.restService = new DekRestService(schemaRegistryUris, timeoutMs, authenticationHeaderValueProvider,
+                    SetSslConfig(config), sslVerify);
+            }
+            else
+            {
+                this.restService = new DekRestService(schemaRegistryUris, timeoutMs, authenticationHeaderValueProvider,
+                    SetSslConfig(config), sslVerify, new X509Certificate2(sslCaLocation));
+            }
         }
 
         /// <summary>
@@ -291,14 +300,6 @@ private List<X509Certificate2> SetSslConfig(IEnumerable<KeyValuePair<string, str
                 certificates.Add(new X509Certificate2(certificateLocation, certificatePassword));
             }
 
-            var caLocation =
-                config.FirstOrDefault(prop => prop.Key.ToLower() == SchemaRegistryConfig.PropertyNames.SslCaLocation)
-                    .Value ?? "";
-            if (!String.IsNullOrEmpty(caLocation))
-            {
-                certificates.Add(new X509Certificate2(caLocation));
-            }
-
             return certificates;
         }
 

src/Confluent.SchemaRegistry.Encryption/Rest/DekRestService.cs

@@ -29,9 +29,9 @@ public class DekRestService : RestService
         /// </summary>
         public DekRestService(string schemaRegistryUrl, int timeoutMs,
             IAuthenticationHeaderValueProvider authenticationHeaderValueProvider, List<X509Certificate2> certificates,
-            bool enableSslCertificateVerification) :
+            bool enableSslCertificateVerification, X509Certificate2 sslCaCertificate = null) :
             base(schemaRegistryUrl, timeoutMs, authenticationHeaderValueProvider, certificates,
-                enableSslCertificateVerification)
+                enableSslCertificateVerification, sslCaCertificate)
         {
         }
 

src/Confluent.SchemaRegistry/CachedSchemaRegistryClient.cs

@@ -355,7 +355,8 @@ public CachedSchemaRegistryClient(IEnumerable<KeyValuePair<string, string>> conf
             if (string.IsNullOrEmpty(sslCaLocation))
             {
                 this.restService = new RestService(schemaRegistryUris, timeoutMs, authenticationHeaderValueProvider, SetSslConfig(config), sslVerify);
-            } else
+            }
+            else
             {
                 this.restService = new RestService(schemaRegistryUris, timeoutMs, authenticationHeaderValueProvider, SetSslConfig(config), sslVerify, new X509Certificate2(sslCaLocation));
             }