Debezium SQL Server连接器SASL_SSL认证超时问题求助

我通过Ansible部署了Confluent Platform 7.1，尝试使用SASL_SSL认证运行Debezium 2.1 SQL Server连接器。未启用认证时连接器运行正常，但启用后出现错误：
org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

日志显示节点已断开，确定是认证问题，尝试了多数公开方案仍未解决。

连接器配置 #

{
  "name": "conn_XXX",
  "config": {
    "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
    "tasks.max": "1",
    "confluent.topic.ssl.endpoint.identification.algorithm": "https",
    "provide.transaction.metadata": "true",
    "errors.retry.timeout": "1000",
    "errors.retry.delay.max.ms": "100",
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topic.prefix": "XXX",
    "database.names": "XXX",
    "database.hostname": "**.**.**.**",
    "database.port": "1433",
    "database.user": "XXX",
    "database.password": "XXX@123",
    "database.instance": "XXXXX",
    "poll.interval.ms": "100",
    "snapshot.mode": "initial",
    "snapshot.include.collection.list": "XXX",
    "snapshot.isolation.mode": "read_committed",
    "signal.data.collection": "XXX.dbo.XXX",
    "table.include.list": "dbo.XXX",
    "schema.history.internal.kafka.topic": "schemahistory_XXX",
    "schema.history.internal.kafka.bootstrap.servers": "broker1.localhost.localdomain:9092,broker2.localhost.localdomain:9092,broker3.localhost.localdomain:9092",
    "sasl.mechanism": "PLAIN",
    "security.protocol": "SASL_SSL",
    "sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka-connect\" password=\"kafka_connect-secret\";",
    "database.encrypt": "false",
    "database.history.kafka.bootstrap.servers": "broker1.localhost.localdomain:9092,broker2.localhost.localdomain:9092,broker3.localhost.localdomain:9092",
    "database.history.consumer.security.protocol": "SASL_SSL",
    "database.history.consumer.sasl.mechanism": "PLAIN",
    "database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka-connect\" password=\"kafka_connect-secret\";",
    "database.history.producer.security.protocol": "SASL_SSL",
    "database.history.producer.sasl.mechanism": "PLAIN",
    "database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka-connect\" password=\"kafka_connect-secret\";",
    "producer.override.sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka-connect\" password=\"kafka_connect-secret\";",
    "producer.override.sasl.mechanism":"PLAIN",
    "producer.override.security.protocol":"SASL_SSL",
    "consumer.override.sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"kafka-connect\" password=\"kafka_connect-secret\";",
    "consumer.override.sasl.mechanism":"PLAIN",
    "consumer.override.security.protocol":"SASL_SSL",
    "connector.client.config.override.policy" : "ALL",
    "database.history.producer.ssl.truststore.location":"/var/ssl/private/kafka_connect.truststore.jks",
    "database.history.producer.ssl.truststore.password":"XXX",
    "database.history.consumer.ssl.truststore.location":"/var/ssl/private/kafka_connect.truststore.jks",
    "database.history.consumer.ssl.truststore.password":"XXX"
  }
}

Ansible TLS相关配置 #

ssl_enabled: true
sasl_protocol: plain
sasl_plain_users:
  admin:
    principal: 'kafka'
    password: 'admin-secret'
  schema_registry:
    principal: 'schema-registry'
    password: 'schema_registry-secret'
  kafka_connect:
    principal: 'kafka-connect'
    password: 'kafka_connect-secret'

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