SQL Server 的 Debezium 问题

Question

我使用了 debezium.io here 提供的教程，并对其进行了调整以使用 MS SQL Server 数据库而不是 MySQL，但是当我在数据库中进行更改时，观察者不会显示任何事件或活动。以下是我采取的步骤：

1. 我运行了 zookeeper docker 命令：

   docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.1

2. 然后我运行了 kafka docker 命令：

   docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:1.1

3. 然后我在启用代理的情况下运行 SQL Server docker 命令：

   docker run -e 'ACCEPT_EULA=Y' -e 'SA_PASSWORD=yourStrong(!)Password' -e 'MSSQL_AGENT_ENABLED=True' --name mssql -p 1433:1433 -d mcr.microsoft.com/mssql/server:2019-latest

4. 然后我连接到 SQL Server 实例并创建了一个名为 PeopleDb 的数据库，并通过运行以下查询创建了一个名为 People 的表：

   USE [PeopleDb] GO

   CREATE TABLE [dbo].[People]([Id] [bigint] IDENTITY(1,1) NOT NULL,[FirstName] [varchar](50) NOT NULL, [LastName] [varchar](50) NOT NULL, CONSTRAINT [PK_People] PRIMARY KEY CLUSTERED ( [Id] ASC)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]) ON [PRIMARY] GO

5. 然后我运行以下命令来启用 CDC：

   Use PeopleDb Go

   EXEC sys.sp_cdc_enable_db

   EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name = N'People', @role_name = Null, @filegroup_name = N'Primary',@supports_net_changes = 0

   EXEC sys.sp_cdc_help_change_data_capture

6. 我注意到 cdc 表是在“系统表”下创建的。当我运行以下查询时，cdc.dbo_People_CT 表中添加了一条记录：

   INSERT INTO TABLE dbo.People(FirstName, LastName) values ('John', 'Smith')

7. 然后我运行了连接器 docker 命令：

   docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mssql:mssql debezium/connect:1.1

8. 然后我通过将以下 Json 发布到 http://localhost:8083/connectors/ 来部署连接器：

   {
       "name": "people-connector",
       "config": {
           "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
           "tasks.max": "1",
           "database.hostname": "mssql",
           "database.port": "1433",
           "database.user": "sa",
           "database.password": "yourStrong(!)Password",
           "database.dbname": "PeopleDb",
           "database.server.id": "184054",
           "database.server.name": "mssql",
           "database.history.kafka.bootstrap.servers": "kafka:9092",
           "database.history.kafka.topic": "mssql.dbo.people",
           "name": "people-connector"
       },
       "tasks": [],
       "type": "source"
   }

9. 我通过检查http://localhost:8083/connectors/people-connector/status 验证了新添加的连接器正在运行：

   {"name":"people-connector","connector":{"state":"RUNNING","worker_id":"172.17.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.17.0.5:8083"}],"type":"source"}

10. 我运行了 watcher docker 命令：

    docker run -it --rm --name watcher --link zookeeper:zookeeper --link kafka:kafka debezium/kafka:1.1 watch-topic -a -k mssql.dbo.people

产生了以下内容：

WARNING: Using default BROKER_ID=1, which is valid only for non-clustered installations.
Using ZOOKEEPER_CONNECT=172.17.0.2:2181
Using KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.17.0.6:9092
Using KAFKA_BROKER=172.17.0.3:9092
Contents of topic mssql.dbo.people:
null    {
  "source" : {
    "server" : "mssql"
  },
  "position" : {
    "transaction_id" : null,
    "event_serial_no" : 1,
    "commit_lsn" : "00000025:000003f8:0003",
    "change_lsn" : "NULL"
  },
  "databaseName" : "PeopleDb",
  "schemaName" : "dbo",
  "tableChanges" : [ {
    "type" : "CREATE",
    "id" : "\"PeopleDb\".\"dbo\".\"People\"",
    "table" : {
      "defaultCharsetName" : null,
      "primaryKeyColumnNames" : [ "Id" ],
      "columns" : [ {
        "name" : "Id",
        "jdbcType" : -5,
        "typeName" : "bigint identity",
        "typeExpression" : "bigint identity",
        "charsetName" : null,
        "length" : 19,
        "scale" : 0,
        "position" : 1,
        "optional" : false,
        "autoIncremented" : false,
        "generated" : false
      }, {
        "name" : "FirstName",
        "jdbcType" : 12,
        "typeName" : "varchar",
        "typeExpression" : "varchar",
        "charsetName" : null,
        "length" : 50,
        "position" : 2,
        "optional" : false,
        "autoIncremented" : false,
        "generated" : false
      }, {
        "name" : "LastName",
        "jdbcType" : 12,
        "typeName" : "varchar",
        "typeExpression" : "varchar",
        "charsetName" : null,
        "length" : 50,
        "position" : 3,
        "optional" : false,
        "autoIncremented" : false,
        "generated" : false
      } ]
    }
  } ]
}

完成这些步骤并确保我有一个正在运行的连接器后，我希望在将新记录插入到 People 表或删除或更新它们时看到新事件，但观察者没有显示任何活动。有谁知道为什么 Debezium 和 SQL Server 之间似乎存在脱节？

Answer 1

根据Debezium SQL Server Connector docs：

SQL Server 连接器将针对单个表的所有插入、更新和删除操作的事件写入单个 Kafka 主题。 Kafka 主题的名称始终采用 serverName.schemaName.tableName 的形式，其中 serverName 是使用 database.server.name 配置属性指定的连接器的逻辑名称， schemaName 是发生操作的模式的名称， tableName 是发生操作的数据库表的名称。

在您的情况下，要查看mssql 数据库中dbo.People 表的更改事件，您需要查看mssql.dbo.People（请注意，Kafka 中的主题名称区分大小写）。