Why Debezium Connector can't connect to a SASL activated broker?

Question

I'm trying to enable SASL/PLAIN for my Kafka system. It works actually, I've tested it for Schema Registry and a Java producer. The problem is Kafka Connect can not establish a connection when SASL is enabled (at least that's what I thought first). I gave the necessary configuration but looks like it doesn't affect at all. I've edited my connect-avro-distributed.properties file as:

sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="secret";

producer.sasl.mechanism=PLAIN
producer.security.protocol=SASL_PLAINTEXT
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="secret";

consumer.sasl.mechanism=PLAIN
consumer.security.protocol=SASL_PLAINTEXT
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="secret";

But in the log it says:

[2022-01-07 12:21:28,645] INFO ProducerConfig values:
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT

Which should be like:

[2022-01-07 12:21:28,645] INFO ProducerConfig values:
        sasl.mechanism = PLAIN
        security.protocol = SASL_PLAINTEXT

Same for the consumer config. What do I need to do? Why it goes with default values? I've restarted the service many times. Thanks in advance.

Edit: There is another connector which runs without any problems and it has correct configuration for SASL.

Edit2: Looks like Debezium connectors need some more configuration in connector side.

Answer 1

Trying it with different connectors made it clear that there was a Debezium specific problem. Since Debezium uses history topics, it needs some additional configuration when security is enabled.

"database.history.consumer.security.protocol": "SASL_PLAINTEXT",
"database.history.consumer.sasl.mechanism": "PLAIN",
"database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"secret\";",
"database.history.producer.security.protocol": "SASL_PLAINTEXT",
"database.history.producer.sasl.mechanism": "PLAIN",
"database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"secret\";"

You need to override default values for both producer and consumer configs of Debezium connector. There are a few lines you need to add if you are using SSL. For more information:https://docs.confluent.io/debezium-connect-sqlserver-source/current/sqlserver_source_connector_config.html

EDIT: For the newer versions of debezium source connector, it should be like this:

"schema.history.internal.consumer.security.protocol": "SASL_PLAINTEXT",
"schema.history.internal.consumer.sasl.mechanism": "PLAIN",
"schema.history.internal.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"secret\";",
"schema.history.internal.producer.security.protocol": "SASL_PLAINTEXT",
"schema.history.internal.producer.sasl.mechanism": "PLAIN",
"schema.history.internal.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"secret\";"