connect to schema registry kafka SSL with flink 1.11.2

[tkg_cangkul]

Hi,

i'm trying to connect to kafka with schema registry that using SSL with 
flink 1.11.2

i've got this error message when i try to submit the job.

Causedby: 
io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 
Unauthorized; error code: 401


Anyone can help me to solve this?

This is my properties :

public static PropertiesgetDefaultProperties() {
    final Properties properties =new Properties(); properties.setProperty("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS); properties.setProperty("security.protocol", KAFKA_SECURITY_PROTOCOL); properties.setProperty("ssl.truststore.location", KAFKA_SSL_TRUSTSTORE_LOCATION); properties.setProperty("ssl.truststore.password", KAFKA_SSL_TRUSTSTORE_PASSWORD); properties.setProperty("ssl.keystore.location", KAFKA_SSL_KEYSTORE_LOCATION); properties.setProperty("ssl.keystore.password", KAFKA_SSL_KEYSTORE_PASSWORD); properties.setProperty("ssl.key.password", KAFKA_SSL_KEY_PASSWORD); // schema registry // properties.setProperty("schema.registry.ssl.truststore.location", KAFKA_SSL_TRUSTSTORE_LOCATION); properties.setProperty("schema.registry.ssl.truststore.password", KAFKA_SSL_TRUSTSTORE_PASSWORD); properties.setProperty("schema.registry.ssl.keystore.location", KAFKA_SSL_KEYSTORE_LOCATION); properties.setProperty("schema.registry.ssl.keystore.password", KAFKA_SSL_KEYSTORE_PASSWORD); properties.setProperty("schema.registry.ssl.key.password", KAFKA_SSL_KEY_PASSWORD); // schema registry end // properties.setProperty("ssl.endpoint.identification.algorithm", ""); properties.setProperty("sasl.mechanism", "PLAIN"); properties.setProperty("basic.auth.credentials.source", "USER_INFO"); properties.setProperty("schema.registry.url", SCHEMA_REGISTRY_URL); properties.setProperty("basic.auth.user.info","user:pass"); properties.setProperty("sasl.jaas.config", 
"org.apache.kafka.common.security.plain.PlainLoginModule required 
username=\"user\"password=\"pass\";"); properties.setProperty("flink.partition-discovery.interval-millis", "60000"); properties.setProperty("request.timeout.ms","60000"); return properties; }