1. 基本實現方案參考SQLServer CDC數據通過Kafka connect實時同步至分析型數據庫 AnalyticDB For PostgreSQL及OSS-阿里云開發(fā)者社區(qū) (aliyun.com)

2. 提示Failed to find any class that implements Connector
   解決方案：假設配置了plugin.path=/home/text/kafka-plugin敬锐，那么在kafka-plugin目錄下建立子目錄sqlserver,然后將插件的所有jar包復制到子目錄中耿战，重啟connect簇秒。

3. 如果kafka通過SASL_PLAINTEXT協議配置了用戶名密碼瞬女，啟動connect的時候，一直報disconnect
   解決方案：需要在connect的配置文件connect-distributed.properties和插件的配置文件里同時添加相關的配置項榕茧。

connect-distributed.properties

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

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

sqlserver-cdc-source.json

{
    "name": "sqlserver-cdc-source",
    "config": {
        "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
        "tasks.max" : "1",
        "database.server.name" : "DESKTOP-0T78TIH",
        "database.hostname" : "192.168.1.1",
        "database.port" : "1433",
        "database.user" : "sa",
        "database.password" : "123456",
        "database.dbname" : "test-cdc",
        "schemas.enable" : "false",
        "mode":"incrementing",
        "incrementing.column.name":"ID",
        "database.history.kafka.bootstrap.servers" : "192.168.1.1:9092",
        "database.history.kafka.topic": "test-cdc.bak",
        "database.history.consumer.sasl.mechanism":"PLAIN",
        "database.history.consumer.sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";",
        "database.history.consumer.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=\"admin\";",
        "database.history.producer.security.protocol":"SASL_PLAINTEXT",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter"

    }
}