title: Getting SSL Exception While Connecting Kafka Cluster with Spring boot and Apache Camel
body: <p>My requirement is to connect the Kafka topic through the SSL with Spring Boot and Apache Camel, for that, I have written the below code but I'm facing an error like Caused by <strong>sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target this</strong></p>
  <p>anyone, please help me how to resolve this error.</p>
  <pre><code>//in this code i'm configured the SSL
      @Configuration
      public class Testing {
          @Bean
           SSLContextParameters sslContextParameters(){
              KeyStoreParameters store = new KeyStoreParameters();
              store.setResource(&quot;kafka.client.truststore.jks&quot;);
              store.setPassword(&quot;123456&quot;);
      
              TrustManagersParameters trust = new TrustManagersParameters();
              trust.setKeyStore(store);
      
              SSLContextParameters parameters = new SSLContextParameters();
              parameters.setTrustManagers(trust);
      
              return parameters;
            }
      
      }
  </code></pre>
  <p>In the below file, I'm calling router with sslContextParameters parameter</p>
  <pre><code>    @Autowired
      SSLContextParameters params;
  @Override
      public void configure() throws Exception {
      from(&quot;{{timerOnce}}&quot;).process(consumerCreate).to(
                  &quot;https://xx.xx.xx.xxx/consumers/group-id?sslContextParameters=params&quot;);
  
  }
  </code></pre>
  <p>****** <em><strong>I've Used Another Approach for Connecting Kafka Cluster through SSL but no luck it's getting exception like this</strong></em>   ****** <em><strong>org.apache.camel.spring.boot.CamelSpringBootInitializationException: java.io.IOException: Invalid Keystore format</strong></em></p>
  <p><em>Below code, I enabled SSL</em></p>
  <pre><code>public Endpoint setupSSLConext(CamelContext camelContext) throws Exception {
  
          KeyStoreParameters keyStoreParameters = new KeyStoreParameters();
          // Change this path to point to your truststore/keystore as jks files
          keyStoreParameters.setResource(&quot;kafka.client.truststore.jks&quot;);
          keyStoreParameters.setPassword(&quot;123456&quot;);
  
          KeyManagersParameters keyManagersParameters = new KeyManagersParameters();
          keyManagersParameters.setKeyStore(keyStoreParameters);
          keyManagersParameters.setKeyPassword(&quot;123456&quot;);
  
          TrustManagersParameters trustManagersParameters = new TrustManagersParameters();
          trustManagersParameters.setKeyStore(keyStoreParameters);
  
          SSLContextParameters sslContextParameters = new SSLContextParameters();
          sslContextParameters.setKeyManagers(keyManagersParameters);
          sslContextParameters.setTrustManagers(trustManagersParameters);
  
          HttpComponent httpComponent = camelContext.getComponent(&quot;https4&quot;, HttpComponent.class);
          httpComponent.setSslContextParameters(sslContextParameters);
  
  
          // This is important to make your cert skip CN/Hostname checks
          httpComponent.setX509HostnameVerifier(new X509HostnameVerifier() {
              @Override
              public void verify(String s, SSLSocket sslSocket) throws IOException {
  
              }
  
              @Override
              public void verify(String s, X509Certificate x509Certificate) throws SSLException {
  
              }
  
              @Override
              public void verify(String s, String[] strings, String[] strings1) throws SSLException {
  
              }
  
              @Override
              public boolean verify(String s, SSLSession sslSession) {
                  // I don't mind just return true for all or you can add your own logic
                  return true;
              }
  
          });
  
          return     httpComponent.createEndpoint(&quot;https://XX.XX.X.XXX/consumers/&quot;);
      }
  </code></pre>
  <p><strong>Below code in the router I used ENDPOINT</strong></p>
  <pre><code>    public void configure() throws Exception {
  
          Endpoint createEndpoint = cdcHelper.setupSSLConext(context);
  
          from(&quot;{{timerOnce}}&quot;).process(consumerCreate)
                  .to(createEndpoint);    // calling kafka consumer 
  
      }
  }
  </code></pre>