Confluent连接器任务失败：Debezium MySQL连接器配置求助

核心错误点 #

从日志可明确定位：Schema History消费者无法连接Kafka Broker（localhost:9092），同时连接器配置存在多处参数错误：

• schema.history.internal.kafka.topic 被错误设置为PLAIN（这是SASL机制名称，而非存储Schema历史的Topic名称）
• SASL JAAS配置不完整，PlainLoginModule未指定用户名和密码
• 数据库历史Topic与业务Topic混用（pageviews）
• Docker环境下，localhost无法指向宿主机的Kafka服务（若Kafka不在Connect容器内）

修正后的连接器配置 #

{
  "name": "MySqlConnectorConnector_0",
  "config": {
    // 数据库历史相关配置
    "database.history.kafka.topic": "test2-db-history",
    "database.history.kafka.bootstrap.servers": "kafka:9092", // 替换为实际Kafka地址，如Docker服务名或宿主机IP
    "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-producer-user\" password=\"producer-pass\";",
    "database.history.producer.ssl.endpoint.identification.algorithm": "https",
    "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-consumer-user\" password=\"consumer-pass\";",
    "database.history.consumer.ssl.endpoint.identification.algorithm": "https",

    // Schema历史相关配置
    "schema.history.internal.kafka.topic": "test2-schema-history",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", // 与上方Kafka地址保持一致

    // 基础数据库配置
    "database.whitelist": "database-test",
    "database.server.name": "database-test",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topic.prefix": "test2",
    "database.hostname": "database-test",
    "database.port": "3306",
    "database.user": "admin",
    "database.password": "**********",
    "database.server.id": "11",
    "database.ssl.mode": "disabled",
    "connect.keep.alive": "true",
    "include.schema.changes": "true",
    "inconsistent.schema.handling.mode": "skip"
  }
}

关键检查项 #

• Kafka地址正确性：Docker环境中，若Kafka与Connect不在同一容器，需使用Docker网络内的服务名（如kafka）或宿主机外部IP，而非localhost
• JAAS配置完整性：确保PlainLoginModule包含username和password参数，注意转义双引号
• Topic命名规范：数据库历史和Schema历史Topic需独立命名，避免与业务Topic冲突
• 网络连通性验证：在Connect容器内执行nc -zv <kafka地址> 9092，确认Kafka Broker可达

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