RDS Postgres error: Database connection failed when reading from copy

[Ervin Hoxha]

Hello,

I would really appreciate any help with something I have been struggling for days.

I am trying to deploy a connector to monitor a RDS Postgres DB. 

Following your tutorial and using the postgres-example image everything works fine, but when I try to connect to a live DB on RDS I start getting errors and seems like the connection fails.

The initial snapshot when I start the connector is performed successfully, but further changes on the DB are not getting tracked because the connector stops (due to these errors).

Restarting the connector does not make any difference.

Some info about my setup:

1. The RDS Postgres instance is version 9.6 and has rds.logical_replication=1 

2. I'm on Mac OS and using the docker compose file from your tutorial to launch all the services:

version: '2'

services:

  zookeeper:

    image: debezium/zookeeper:0.7.5

    ports:

     - 2181:2181

     - 2888:2888

     - 3888:3888

  kafka:

    image: debezium/kafka:0.7.5

    ports:

     - 9092:9092

    links:

     - zookeeper

    environment:

     - ZOOKEEPER_CONNECT=zookeeper:2181

  postgres:

    image: debezium/example-postgres:0.8

    ports:

     - 5432:5432

    environment:

     - POSTGRES_USER=postgres

     - POSTGRES_PASSWORD=postgres

  connect:

    image: debezium/connect:0.7.5

    ports:

     - 8083:8083

    links:

     - kafka

     - postgres

    environment:

     - BOOTSTRAP_SERVERS=kafka:9092

     - GROUP_ID=1

     - CONFIG_STORAGE_TOPIC=my_connect_configs

     - OFFSET_STORAGE_TOPIC=my_connect_offsets

3. Ad I mentioned above, all tests with the example-postgres image are working fine

4. Next step I tried to deploy a connector for my RDS instance

Note:  tried both "wal2json" and "wal2json_rds" plugins, with the same result. 

Also tried without the "slot.drop_on_stop" option

{

    "name": "my-connector",

    "config": {

        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

        "tasks.max": "1",

        "database.hostname": "xxxxxx.cg7ckvedrvzr.eu-west-1.rds.amazonaws.com",

        "database.port": "5432",

        "database.user": "pguser",

        "database.password": "pgpass",

        "database.dbname" : "pgdbname",

        "database.server.name": "pgdbname",

        "database.whitelist": "pgdbname",

        "database.history.kafka.bootstrap.servers": "kafka:9092",

        "database.history.kafka.topic": "schema-changes.inventory",

        "plugin.name": "wal2json_rds",

        "table.whitelist": "public.campaign",

        "slot.drop_on_stop":false

    }

}

After the connector starts and performs the first snapshot successfully I'm facing these errors:

Error from Kafka Connect:

connect_1    | 2018-05-18 08:58:44,909 INFO   Postgres|optimizer_v3|records-stream-producer  REPLICA IDENTITY for 'public.campaign' is 'DEFAULT'; UPDATE and DELETE events will contain previous values only for PK columns   [io.debezium.connector.postgresql.PostgresSchema]

connect_1    | 2018-05-18 08:58:44,911 INFO   Postgres|optimizer_v3|records-stream-producer  Creating thread debezium-postgresconnector-optimizer_v3-records-stream-producer   [io.debezium.util.Threads]

connect_1    | 2018-05-18 08:58:44,912 INFO   ||  WorkerSourceTask{id=native-optimizer-connector-0} Source task finished initialization and start   [org.apache.kafka.connect.runtime.WorkerSourceTask]

connect_1    | 2018-05-18 08:58:45,008 WARN   Postgres|optimizer_v3|records-stream-producer  Closing replication stream due to db connection IO exception...   [io.debezium.connector.postgresql.RecordsStreamProducer]

connect_1    | 2018-05-18 08:58:45,448 INFO   ||  WorkerSourceTask{id=native-optimizer-connector-0} Committing offsets   [org.apache.kafka.connect.runtime.WorkerSourceTask]

connect_1    | 2018-05-18 08:58:45,448 INFO   ||  WorkerSourceTask{id=native-optimizer-connector-0} flushing 1 outstanding messages for offset commit   [org.apache.kafka.connect.runtime.WorkerSourceTask]

connect_1    | 2018-05-18 08:58:45,471 INFO   ||  WorkerSourceTask{id=native-optimizer-connector-0} Finished commitOffsets successfully in 23 ms   [org.apache.kafka.connect.runtime.WorkerSourceTask]

connect_1    | 2018-05-18 08:58:45,472 ERROR  ||  WorkerSourceTask{id=native-optimizer-connector-0} Task threw an uncaught and unrecoverable exception   [org.apache.kafka.connect.runtime.WorkerTask]

connect_1    | org.apache.kafka.connect.errors.ConnectException: An exception ocurred in the change event producer. This connector will be stopped.

connect_1    | at io.debezium.connector.base.ChangeEventQueue.throwProducerFailureIfPresent(ChangeEventQueue.java:168)

connect_1    | at io.debezium.connector.base.ChangeEventQueue.poll(ChangeEventQueue.java:149)

connect_1    | at io.debezium.connector.postgresql.PostgresConnectorTask.poll(PostgresConnectorTask.java:144)

connect_1    | at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:179)

connect_1    | at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)

connect_1    | at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)

connect_1    | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)

connect_1    | at java.util.concurrent.FutureTask.run(FutureTask.java:266)

connect_1    | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

connect_1    | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

connect_1    | at java.lang.Thread.run(Thread.java:748)

connect_1    | Caused by: org.postgresql.util.PSQLException: Database connection failed when reading from copy

connect_1    | at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:964)

connect_1    | at org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)

connect_1    | at org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:145)

connect_1    | at org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:114)

connect_1    | at org.postgresql.core.v3.replication.V3PGReplicationStream.read(V3PGReplicationStream.java:60)

connect_1    | at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.read(PostgresReplicationConnection.java:189)

connect_1    | at io.debezium.connector.postgresql.RecordsStreamProducer.streamChanges(RecordsStreamProducer.java:119)

connect_1    | at io.debezium.connector.postgresql.RecordsStreamProducer.lambda$start$0(RecordsStreamProducer.java:105)

connect_1    | ... 5 more

connect_1    | Caused by: java.io.EOFException

connect_1    | at org.postgresql.core.PGStream.receiveChar(PGStream.java:284)

connect_1    | at org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1006)

connect_1    | at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:962)

connect_1    | ... 12 more

connect_1    | 2018-05-18 08:58:45,472 ERROR  ||  WorkerSourceTask{id=native-optimizer-connector-0} Task is being killed and will not recover until manually restarted   [org.apache.kafka.connect.runtime.WorkerTask]

Connector status:

{

   "name":"my-connector",

   "connector":{

      "state":"RUNNING",

      "worker_id":"172.20.0.5:8083"

   },

   "tasks":[

      {

         "state":"FAILED",

         "trace":"org.apache.kafka.connect.errors.ConnectException: An exception ocurred in the change event producer. This connector will be stopped.\n\tat io.debezium.connector.base.ChangeEventQueue.throwProducerFailureIfPresent(ChangeEventQueue.java:168)\n\tat io.debezium.connector.base.ChangeEventQueue.poll(ChangeEventQueue.java:149)\n\tat io.debezium.connector.postgresql.PostgresConnectorTask.poll(PostgresConnectorTask.java:144)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:179)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)\n\tat java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:748)\nCaused by: org.postgresql.util.PSQLException: Database connection failed when reading from copy\n\tat org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:964)\n\tat org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)\n\tat org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:145)\n\tat org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:114)\n\tat org.postgresql.core.v3.replication.V3PGReplicationStream.read(V3PGReplicationStream.java:60)\n\tat io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.read(PostgresReplicationConnection.java:189)\n\tat io.debezium.connector.postgresql.RecordsStreamProducer.streamChanges(RecordsStreamProducer.java:119)\n\tat io.debezium.connector.postgresql.RecordsStreamProducer.lambda$start$0(RecordsStreamProducer.java:105)\n\t... 5 more\nCaused by: java.io.EOFException\n\tat org.postgresql.core.PGStream.receiveChar(PGStream.java:284)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1006)\n\tat org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:962)\n\t... 12 more\n",

         "id":0,

         "worker_id":"172.20.0.5:8083"

      }

   ],

   "type":"source"

}

Thank you for the help and sorry for this long post.

Ervin

[hothu...@unimarmores.com]

I have the same problem, did you have the answer?

[himanshu sawant]

Me too...any update on this??

[Jiri Pechanec]

Always check if there is any message in database log too.

J.

On Monday, June 10, 2019 at 10:34:59 PM UTC+2, himanshu sawant wrote:

Me too...any update on this??