Schema Registry configuration provided through key/value converter in a connector are not substituted by a ConfigProvider

[yoniwg]

I am using Kafka-Connect on Kubernetes using the EnvVarConfigProvider:

    config.providers: env
    config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider

And provided all Schema Registry configuration through env-vars to the workers

SCHEMA_REGISTRY_URL
SCHEMA_REGISTRY_TRUSTSTORE_LOCATION
SCHEMA_REGISTRY_TRUSTSTORE_PASSWORD
SCHEMA_REGISTRY_BASIC_AUTH_CREDENTIALS_SOURCE
SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO

When using them in the connector configuration itself they work just fine, for example with JDBC sink connector:

{
   "connector.class":"com.securithings.sp.connect.jdbc.JdbcBatchSinkConnector", 
   
   "schema.registry.url": "${env:SCHEMA_REGISTRY_URL}",
   "schema.registry.ssl.truststore.location": "${env:SCHEMA_REGISTRY_TRUSTSTORE_LOCATION}",
    "schema.registry.ssl.truststore.password": "${env:SCHEMA_REGISTRY_TRUSTSTORE_PASSWORD}",
    "schema.registry.basic.auth.credentials.source": "${env:SCHEMA_REGISTRY_BASIC_AUTH_CREDENTIALS_SOURCE}",
    "schema.registry.basic.auth.user.info": "${env:SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO}",
   ...
}

But when I assign them to the converters configuration, for example (for a source connector):

{
    ...
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "${env:SCHEMA_REGISTRY_URL}",
    "key.converter.schema.registry.ssl.truststore.location": "${env:SCHEMA_REGISTRY_TRUSTSTORE_LOCATION}",
    "key.converter.schema.registry.ssl.truststore.password": "${env:SCHEMA_REGISTRY_TRUSTSTORE_PASSWORD}",
    "key.converter.basic.auth.credentials.source": "${env:SCHEMA_REGISTRY_BASIC_AUTH_CREDENTIALS_SOURCE}",
    "key.converter.basic.auth.user.info": "${env:SCHEMA_REGISTRY_BASIC_AUTH_USER_INFO}",
    "key.converter.auto.register.schemas": "false",
    ...
}

All of the properties which start with key.converter.schema.registry (except for the url) are not substitute by the env-vars.

I dug into the code, and it seems that the creation of the schema-registry in the Avro converter, simply uses the config as a Map, and does not use the ConfigProvider to reslove the properties. See here:

schema-registry/avro-converter/src/main/java/io/confluent/connect/avro/AvroConverter.java

Line 71 in 99fc46b

schemaRegistry = SchemaRegistryClientFactory.newClient(