我正在尝试让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
修好了。
我自己的愚蠢错误。我使用的是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个处理器的配置文件: 现在我启动处理器: 类型元组包含配置文