如何使用debezium从Postgres流式传输更改

Question

使用debezium从Postgres流式更改

已完成的设置:

1. Docker设置。
   
2. 启动Postgres、zookeeper、kafka和debezium Connector。
   
3. 使用decoderbufs、wal2json(postgres)设置远程数据库。
   
4. 使用curl连接到debezium。
   
5. 创建了一个观察者。
   

问题：当我启动watcher时，它读取了之前发生的所有更改，但当任何插入完成时，kafka会向debezium抛出一个异常，表示“在change event Producer.This connector will be stopped.”，并且在watcher中没有显示。

由于我对这些概念很陌生，不知道我在环境设置中错过了什么，这是我在堆栈溢出中的第一个问题，请忽略我的错误。

主要问题是我的本地数据库运行良好。
有没有人能帮上忙？
请提前致谢

019-05-02 14:09:47,242 WARN   Postgres|kafkaserver|records-stream-producer  Closing replication stream due to db connection IO exception...   [io.debezium.connector.postgresql.RecordsStreamProducer]
2019-05-02 14:09:47,365 INFO   ||  WorkerSourceTask{id=kafka-public-connector-0} Committing offsets   [org.apache.kafka.connect.runtime.WorkerSourceTask]
2019-05-02 14:09:47,366 INFO   ||  WorkerSourceTask{id=kafka-public-connector-0} flushing 0 outstanding messages for offset commit   [org.apache.kafka.connect.runtime.WorkerSourceTask]
2019-05-02 14:09:47,375 ERROR  ||  WorkerSourceTask{id=kafka-public-connector-0} Task threw an uncaught and unrecoverable exception   [org.apache.kafka.connect.runtime.WorkerTask]
org.apache.kafka.connect.errors.ConnectException: An exception ocurred in the change event producer. This connector will be stopped.
    at io.debezium.connector.base.ChangeEventQueue.throwProducerFailureIfPresent(ChangeEventQueue.java:170)
    at io.debezium.connector.base.ChangeEventQueue.poll(ChangeEventQueue.java:151)
    at io.debezium.connector.postgresql.PostgresConnectorTask.poll(PostgresConnectorTask.java:161)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:245)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:221)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.postgresql.util.PSQLException: Database connection failed when reading from copy
    at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1037)
    at org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)
    at org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:155)
    at org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:124)
    at org.postgresql.core.v3.replication.V3PGReplicationStream.read(V3PGReplicationStream.java:70)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.read(PostgresReplicationConnection.java:251)
    at io.debezium.connector.postgresql.RecordsStreamProducer.streamChanges(RecordsStreamProducer.java:134)
    at io.debezium.connector.postgresql.RecordsStreamProducer.lambda$start$0(RecordsStreamProducer.java:120)
    ... 5 more
Caused by: java.io.EOFException
    at org.postgresql.core.PGStream.receiveChar(PGStream.java:308)
    at org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1079)
    at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1035)
    ... 12 more
2019-05-02 14:09:47,387 ERROR  ||  WorkerSourceTask{id=kafka-public-connector-0} Task is being killed and will not recover until manually restarted   [org.apache.kafka.connect.runtime.WorkerTask]

• 是否有解决此问题的完整指南？
• 的主要目标是，我有一个数据库中有大量的数据，其中一个应用程序（生产者）从另一个服务器获取数据，并将所有数据保存在我们自己的数据库中，另一个应用程序（消费者）获取存储并应用业务逻辑和前端。在这里，我想用这个debezium和kafka部分替换从另一个应用程序（消费者）到db的点击。
• 或者有什么方法可以这样做。

Answer 1

感谢大家。解决了上述问题。远程数据库中的安装存在实际问题。很少有其他依赖项没有正确安装，如postgis、protobuf-c、decoderbufs，一旦正确安装，问题就解决了。