Debezium连接器报`Unexpected Kafka request of type METADATA during SASL handshake`错误求助

我们已通过以下producer.config/consumer.config成功部署Kafka Producer/Consumer（临时密码，无需在意）：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="user1" \
    password="JXtC77mME4";

随后尝试通过以下HTTP请求创建Debezium SQL Server连接器：

{
    "name": "sql_connector",
    "config": {
        "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
        "database.hostname": "REMOVED.domain.local",
        "database.port": "1764",
        "database.user": "REMOVED-user",
        "database.password": "REMOVED-password",
        "database.names": "MM",
        "database.applicationIntent": "ReadOnly",
        "snapshot.isolation.mode": "snapshot",
        "snapshot.mode": "initial",
        "message.key.columns": "MM.DB_TABLE:rowid",
        "topic.prefix": "mmv2_mssqlami",
        "table.include.list": "dbo.DB_TABLE",
        "schema.history.internal.kafka.bootstrap.servers": "kafka.ky9998-003.svc.cluster.local:9092",
        "schema.history.internal.kafka.topic": "mmv2_amimssql",
        "schema.history.internal.consumer.security.protocol": "SASL_PLAINTEXT",
        "schema.history.internal.consumer.sasl.mechanism": "SCRAM-SHA-256",
        "schema.history.internal.consumer.sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"user1\" password=\"JXtC77mME4\";",
        "schema.history.internal.producer.security.protocol": "SASL_PLAINTEXT",
        "schema.history.internal.producer.sasl.mechanism": "SCRAM-SHA-256",
        "schema.history.internal.producer.sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"user1\" password=\"JXtC77mME4\";",
        "database.encrypt": "false",
        "slot.name":"test1",
        "key.converter.schemas.enable":"false",
        "value.converter.schemas.enable":"false",
        "key.converter":"org.apache.kafka.connect.json.JsonConverter",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter",
        "decimal.handling.mode": "string",
        "datetime.handling.mode": "string",
        "tombstones.on.delete": "false",
        "group.id": "debezium-mssql-group"
        // ... transforms removed
    }
}

执行后，Kafka Broker日志反复出现以下错误：

[2025-03-12 00:44:20,691] INFO [SocketServer listenerType=BROKER, nodeId=0] Failed authentication with /10.1.201.176 (channelId=10.1.33.149:9092-10.1.201.176:35618-61) (Unexpected Kafka request of type METADATA during SASL handshake.) (org.apache.kafka.common.network.Selector)

错误原因 #

这个错误的核心问题是：Kafka Connect工作进程本身没有配置SASL认证参数。你只给Debezium内部用于存储schema历史的Producer/Consumer配置了安全参数，但Connect作为独立服务，它自身与Kafka集群的连接（比如发送CDC数据到目标Topic、管理消费偏移量等）并未启用SASL认证，导致它跳过SASL握手直接向Broker发送METADATA请求，被Broker拒绝。

修正方案 #

1. 添加Kafka Connect全局SASL配置
   找到Connect的配置文件（通常是connect-distributed.properties或connect-standalone.properties），加入以下认证参数：

   security.protocol=SASL_PLAINTEXT
   sasl.mechanism=SCRAM-SHA-256
   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
       username="user1" \
       password="JXtC77mME4";
   

   如果是容器部署，可通过环境变量传递配置：

   • KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT
   • KAFKA_SASL_MECHANISM=SCRAM-SHA-256
   • KAFKA_SASL_JAAS_CONFIG=org.apache.kafka.common.security.scram.ScramLoginModule required username="user1" password="JXtC77mME4";

2. 重启Kafka Connect服务
   修改全局配置后，必须重启Connect进程才能让新的安全配置生效。

3. 重新提交连接器配置
   重启完成后，重新提交之前的Debezium连接器配置，此时Connect会按照SASL流程与Broker建立连接，错误日志将不再出现。

内容的提问来源于stack exchange，提问作者user1913559