当前位置: 首页 > 知识库问答 >
问题:

Debezium SQLServerConnector for Kakfa Connect未在Kafka中创建主题

田瀚
2023-03-14

我正在尝试让Debezium的SQLServerConnector在Kafka Connect中工作,除了没有创建主题外,所有的东西似乎都运行得很好。我是不是漏掉了什么?

它将使用正确的配置启动任务...

Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,424] INFO Kafka version : 2.1.1-cp1 (org.apache.kafka.common.utils.AppInfoParser:109)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,424] INFO Kafka commitId : f5b753880d5460f1 (org.apache.kafka.common.utils.AppInfoParser:110)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,425] INFO Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:860)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,426] INFO Starting SqlServerConnectorTask with configuration: (io.debezium.connector.common.BaseSourceTask:42)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,426] INFO    connector.class = io.debezium.connector.sqlserver.SqlServerConnector (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,426] INFO    database.user = cdc_test (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,426] INFO    database.dbname = kafka_sql_test (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,426] INFO    task.class = io.debezium.connector.sqlserver.SqlServerConnectorTask (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.hostname = mbvmsqlsvr001-t (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.history.kafka.bootstrap.servers = horton1:9092,horton2:9092,horton3:9092,horton4:9092 (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.history.kafka.topic = dbhistory.sql2016test (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.password = ******** (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    name = sql-server-test-connector (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.server.name = mbvmsqlsvr001-t (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    database.port = 1433 (io.debezium.connector.common.BaseSourceTask:44)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,427] INFO    table.whitelist = dbo.message_table (io.debezium.connector.common.BaseSourceTask:44)

获取快照设置......

Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,430] INFO Kafka version : 2.1.1-cp1 (org.apache.kafka.common.utils.AppInfoParser:109)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,430] INFO Kafka commitId : f5b753880d5460f1 (org.apache.kafka.common.utils.AppInfoParser:110)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,522] INFO Requested thread factory for connector SqlServerConnector, id = mbvmsqlsvr001-t named = change-event-source-coordinator (io.debezium.util.Threads:23
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,523] INFO Creating thread debezium-sqlserverconnector-mbvmsqlsvr001-t-change-event-source-coordinator (io.debezium.util.Threads:247)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,523] INFO WorkerSourceTask{id=sql-server-test-connector-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:19
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,523] INFO No previous offset has been found (io.debezium.connector.sqlserver.SqlServerSnapshotChangeEventSource:56)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,523] INFO According to the connector configuration both schema and data will be snapshotted (io.debezium.connector.sqlserver.SqlServerSnapshotChangeEventSourc
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,523] INFO Snapshot step 1 - Preparing (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:85)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,524] INFO Snapshot step 2 - Determining captured tables (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:94)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,528] INFO Snapshot step 3 - Locking captured tables (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:100)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,530] INFO Schema locking was disabled in connector configuration (io.debezium.connector.sqlserver.SqlServerSnapshotChangeEventSource:84)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,530] INFO Snapshot step 4 - Determining snapshot offset (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:106)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,530] INFO Snapshot step 5 - Reading structure of captured tables (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:109)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,530] INFO Snapshot step 6 - Persisting schema history (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:113)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,531] INFO Snapshot step 7 - Snapshotting data (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:125)
Apr 16 15:30:00 horton5 connect-distributed[9469]: [2019-04-16 15:30:00,531] INFO Snapshot step 8 - Finalizing (io.debezium.relational.HistorizedRelationalSnapshotChangeEventSource:149)

并且连接器任务正在运行...

# curl  http://172.24.32.10:8083/connectors/sql-server-test-connector/status | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   175  100   175    0     0  73160      0 --:--:-- --:--:-- --:--:-- 87500
{
  "name": "sql-server-test-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "horton5:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "horton5:8083"
    }
  ],
  "type": "source"
}

用户名和密码在SQL Server上有一个SPID,并运行CDC查询以获取更改,但在Kafka主题中没有显示任何内容。

我有什么不明白的?我是否应该先创建主题,然后如何配置连接器以向其发布消息?

更新:这里是连接器json

    curl -X POST -H "Content-Type: application/json" --data '{
  "name": "sql-server-test-connector",
  "config": {
    "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
    "database.hostname": "mbvmsqlsvr001-t",
    "database.port": "1433",
    "database.user": "cdc_test",
    "database.password": "Password!123",
    "database.dbname": "kafka_sql_test",
    "database.server.name": "mbvmsqlsvr001-t",
    "table.whitelist": "dbo.message_table",
    "database.history.kafka.bootstrap.servers": "horton1:9092,horton2:9092,horton3:9092,horton4:9092",
    "database.history.kafka.topic": "dbhistory.sql2016test"
    }
  }' http://172.24.32.10:8083/connectors

共有1个答案

王高邈
2023-03-14

修好了。

我自己的愚蠢错误。我使用的是debezium-connector-sqlserver-0.9.0.alpha1-plugin.tar.gz发行版。我将debezium-connector-sqlserver-0.9.4.final-plugin.tar.gz放在适当的位置,一切都运行得很好(我可以几乎实时地看到消费者的变化)。

我都不知道自己是怎么拿到阿尔法版的......

 类似资料:
  • 使用命令时: /usr/local/kafka/bin/kafka-topics.sh--创建--zookeeper localhost:2181--复制-因子1--分区1--主题测试 提前道谢。

  • 问题内容: 我想通过Java在Kafka(kafka_2.8.0-0.8.1.1)中创建一个主题。如果我在命令提示符下创建一个主题,并且如果我通过java api推送消息,它也可以正常工作。但是我想通过java api创建一个主题。经过长时间的搜索,我发现了以下代码, 我尝试了上面的代码,它表明创建了主题,但是无法在该主题中推送消息。我的代码有什么问题吗?还是通过其他方式实现以上目标? 问题答案:

  • 1)我在机器上启动Kafka 2)我用配置启动Spring Boot服务器: 1)为什么会发生这种情况? 2)我必须创建吗?为什么不需要本地Kafka? 埃德迪特

  • 我要用Kafka流计算平均值。所以我做了一个有状态的操作,聚合,需要创建一个状态存储,但是这种情况不会发生。 这里是平均值的函数: 以下是例外情况: 问题是基本目录不存在,但我希望kafka流在必要时创建目录。 编辑 ----- 我注意到,如果我有1个处理器,使一个变量的平均值没有问题,但如果我有2个处理器是。 1个处理器的配置文件: 2个处理器的配置文件: 现在我启动处理器: 类型元组包含配置文