public SchemaRegistryClient get() {
    if (schemaRegistryUrl.equals("")) {
      return new DefaultSchemaRegistryClient();
    }
  
    final RestService restService = serviceSupplier.get();
    // This call sets a default sslSocketFactory.
    final SchemaRegistryClient client = schemaRegistryClientFactory.create(
        restService,
        1000,
        ImmutableList.of(
            new AvroSchemaProvider(), new ProtobufSchemaProvider(), new JsonSchemaProvider()),
        schemaRegistryClientConfigs,
        httpHeaders
    );

    // If we have an sslContext, we use it to set the sslSocketFactory on the restService client.
    //  We need to do it in this order so that the override here is not reset by the constructor
    //  above.
    if (sslContext != null) {
      restService.setSslSocketFactory(sslContext.getSocketFactory());
    }
    return client;
  }
@Test
  public void shouldPassBasicAuthCredentialsToSchemaRegistryClient() {
    // Given
    final Map<String, Object> schemaRegistryClientConfigs = ImmutableMap.of(
        "ksql.schema.registry.basic.auth.credentials.source", "USER_INFO",
        "ksql.schema.registry.basic.auth.user.info", "username:password",
        "ksql.schema.registry.url", "some url"
    );

    final KsqlConfig config = new KsqlConfig(schemaRegistryClientConfigs);

    final Map<String, Object> expectedConfigs = defaultConfigs();
    expectedConfigs.put("basic.auth.credentials.source", "USER_INFO");
    expectedConfigs.put("basic.auth.user.info", "username:password");

    // When:
    new KsqlSchemaRegistryClientFactory(
        config, restServiceSupplier, SSL_CONTEXT, srClientFactory, Collections.emptyMap()).get();

    // Then:
    verify(restService).setSslSocketFactory(isA(SSL_CONTEXT.getSocketFactory().getClass()));
    srClientFactory.create(same(restService), anyInt(), any(), eq(expectedConfigs), any());
  }