Debezium / kafka 连接既不创建主题也不将数据推送到创建的主题

Question

我有 zookeeper、kafka 和 kafka connect 使用 debezium 插件运行。 我看不到任何错误。 我已将一个数据库和几个表列入白名单。 在插入或更新表时，不会创建主题； /tmp/kafka-logs 中的主题没有数据目录。 我刚刚尝试为其中一个表创建一个主题（一种带有一列的虚拟表 't'），但是在将一行插入 t 后，该主题的日志中没有数据......日志截图如下

只有一个主题（我创建的那个，看看如果我自己创建它是否会被填充；我没想到会这样，但事实并非如此）：

vagrant@coton:~$ ./kafka/kafka_2.13-2.8.0/bin/kafka-topics.sh --bootstrap-server 1.2.3.4:9092 --list
vagrant.bet.t
vagrant@coton:~$ 

连接器状态为运行：

vagrant@coton:~$ curl -X GET http://localhost:8083/connectors/pg-sport-connector/ | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   940  100   940    0     0  50165      0 --:--:-- --:--:-- --:--:-- 52222
{
  "name": "pg-sport-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.user": "postgres",
    "database.dbname": "sport",
    "transforms": "unwrap",
    "database.server.name": "vagrant",
    "database.port": "5432",
    "plugin.name": "pgoutput",
    "table.whitelist": "bet.event,bet.t",
    "internal.key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "decimal.handling.mode": "string",
    "database.hostname": "localhost",
    "database.password": "",
    "value.converter.schemas.enable": "false",
    "internal.value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "name": "pg-sport-connector",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "database.whitelist": "sport",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter"
  },
  "tasks": [
    {
      "connector": "pg-sport-connector",
      "task": 0
    }
  ],
  "type": "source"
}

vagrant@coton:~$ curl -X GET http://localhost:8083/connectors/pg-sport-connector/status | jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   172  100   172    0     0  10876      0 --:--:-- --:--:-- --:--:-- 11466
{
  "name": "pg-sport-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "127.0.1.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "127.0.1.1:8083"
    }
  ],
  "type": "source"
}

运行 Java 8。谢谢。

Answer 1

尝试在你的配置中添加这个

"database.history.kafka.bootstrap.servers": "kafka:9092",  
"database.history.kafka.topic": "schema-changes.your_topic_name"

然后用这个检查你的主题名称

./kafka/kafka_2.13-2.8.0/bin/kafka-topics.sh --zookeeper localhost:2181 --list

在此之后，您可以检查表格中的任何更改

./kafka/kafka_2.13-2.8.0/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic your_topic_name