为 Kafka 和 MySQL 使用 Debezium 连接器时出错

Question

我在远程服务器中使用 Kafka Debezium MySQL 连接器。 我不断收到两条错误消息，它们随机出现（我没有编辑任何内容）。

第一个是

[2021-02-26 14:08:00,523] INFO Kafka commitId: 9c1fbb3db1e0d69d (org.apache.kafka.common.utils.AppInfoParser:118)
[2021-02-26 14:08:00,523] INFO Kafka startTimeMs: 1614370080522 (org.apache.kafka.common.utils.AppInfoParser:119)
[2021-02-26 14:08:00,542] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:00,645] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:00,846] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:01,148] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:01,550] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:02,353] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)
[2021-02-26 14:08:03,358] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient:757)

第二个是

[2021-02-26 14:05:07,241] INFO Connected to MySQL binlog at localhost:3306, starting at GTIDs bb53ae0b-65d2-11eb-abc2-00505685c6cc:1-107 and binlog file 'binlog.000032', pos=25915, skipping 0 events plus 0 rows (io.debezium.connector.mysql.BinlogReader:1128)
[2021-02-26 14:05:07,241] INFO Waiting for keepalive thread to start (io.debezium.connector.mysql.BinlogReader:415)
[2021-02-26 14:05:07,242] INFO Creating thread debezium-mysqlconnector-OutboxDemo3-binlog-client (io.debezium.util.Threads:287)
[2021-02-26 14:05:07,342] INFO Keepalive thread is running (io.debezium.connector.mysql.BinlogReader:422)
[2021-02-26 14:05:07,442]ERROR Failed to send HTTP request to endpoint: http://localhost:8081/subjects/OutboxDemo3-key/versions (io.confluent.kafka.schemaregistry.client.rest.RestService:267)
java.net.ConnectException: Connection refused (Connection refused)
    at java.base/java.net.PlainSocketImpl.socketConnect(Native Method)
...
    at java.base/java.lang.Thread.run(Thread.java:834)
[2021-02-26 14:05:07,513] INFO WorkerSourceTask{id=outbox-location-connector3-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:481)
[2021-02-26 14:05:07,513] INFO WorkerSourceTask{id=outbox-location-connector3-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:498)
[2021-02-26 14:05:07,513] ERROR WorkerSourceTask{id=outbox-location-connector3-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:187)
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:311)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:340)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:264)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.connect.errors.DataException: Failed to serialize Avro data from topic OutboxDemo3 :
    at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:91)
    at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$1(WorkerSourceTask.java:311)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
    ... 11 more
Caused by: org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
Caused by: java.net.ConnectException: Connection refused (Connection refused)
    at java.base/java.net.PlainSocketImpl.socketConnect(Native Method)
    at java.base/java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:399)
    at java.base/java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:242)
    at java.base/java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:224)
    at java.base/java.net.Socket.connect(Socket.java:609)

有人可以帮忙吗？非常感谢！

Answer 1

这个问题是因为架构注册表没有正确启动。在我编辑 https://github.com/confluentinc/confluent-cli/issues/120 之后的代理属性并手动启动模式注册表后，连接器可以工作。

Answer 2

为了解决这个问题，请配置接受旧加密算法密码的MySQL 8.0服务器。

• MySQL 8.0 对密码使用安全密码认证算法。
• 旧版本的 MySQL 使用 sha2_password 认证算法。 在跳转修复它之前，我们需要知道我们当前用户的名字。然后，我们可以使用以下命令：

mysql> select current_user();
+----------------+
| current_user() |
+----------------+
| root@%         |
+----------------+
1 row in set (0.03 sec)

这是您需要使用密码运行的命令。

ALTER USER 'root'@'%' IDENTIFIED WITH mysql_native_password BY 'our_password';