title: How to access a confluent schema registry server secured with a password using Spring cloud stream?
body: <p>I'm using spring cloud stream alongside <a href="https://aiven.io/kafka" rel="nofollow noreferrer">Aiven</a>'s schema registry which  uses <a href="https://github.com/confluentinc/schema-registry" rel="nofollow noreferrer">confluent's schema registry</a>. Aiven's schema registry is secured with a password. Based on <a href="https://help.aiven.io/kafka/kafka-add-ons/using-schema-registry-with-aiven-kafka" rel="nofollow noreferrer">these</a> instructions, these two config parameters need to be set to successfully access the schema registry server.</p>
  
  <pre><code> props.put("basic.auth.credentials.source", "USER_INFO");
   props.put("basic.auth.user.info", "avnadmin:schema-reg-password");
  </code></pre>
  
  <p>Everything is fine when I only use vanilla java's kafka drivers, but if I use Spring cloud stream, I don't know how to inject these two parameters. At the moment, I'm putting <code>"basic.auth.user.info"</code> and <code>"basic.auth.credentials.source"</code> under <code>"spring.cloud.stream.kafka.binder.configuration"</code> in the <code>application.yml</code> file.</p>
  
  <p>Doing this, I'm getting <code>"401 Unauthorized"</code> on the line where the schema wants to get registered. </p>
  
  <p>Update 1:</p>
  
  <p>Based on 'Ali n's suggestion, I updated the way SchemaRegistryClient's bean was configured so that it becomes aware of the SSL context.</p>
  
  <pre><code>@Bean
  public SchemaRegistryClient schemaRegistryClient(
      @Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint) {
    try {
      final KeyStore keyStore = KeyStore.getInstance("PKCS12");
      keyStore.load(new FileInputStream(
              new File("path/to/client.keystore.p12")),
          "secret".toCharArray());
  
      final KeyStore trustStore = KeyStore.getInstance("JKS");
      trustStore.load(new FileInputStream(
              new File("path/to/client.truststore.jks")),
          "secret".toCharArray());
  
      TrustStrategy acceptingTrustStrategy = (X509Certificate[] chain, String authType) -&gt; true;
  
      SSLContext sslContext = SSLContextBuilder
          .create()
          .loadKeyMaterial(keyStore, "secret".toCharArray())
          .loadTrustMaterial(trustStore, acceptingTrustStrategy)
          .build();
  
      HttpClient httpClient = HttpClients.custom().setSSLContext(sslContext).build();
      ClientHttpRequestFactory requestFactory = new HttpComponentsClientHttpRequestFactory(
          httpClient);
      ConfluentSchemaRegistryClient schemaRegistryClient = new ConfluentSchemaRegistryClient(
          new RestTemplate(requestFactory));
      schemaRegistryClient.setEndpoint(endpoint);
      return schemaRegistryClient;
    } catch (Exception ex) {
      ex.printStackTrace();
      return null;
    }
  }
  </code></pre>
  
  <p>This helped getting rid of the error on app's startup and registered the schema. However, whenever the app wanted to push a message to Kafka, a new error was thrown again. Finally this was also fixed by mmelsen's answer.</p>