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.
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\";"