Debezium Connector for PostgreSQL

[Igor Felix]

Hello everyone,

Can somebody help me? I am trying to start a worker using Debezium Connector for PostgreSQL and I am getting the following error:

[2017-05-11 09:30:28,753] ERROR Task test-connector-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:141)
java.lang.NoSuchMethodError: org.postgresql.PGConnection.getReplicationAPI()Lorg/postgresql/replication/PGReplicationConnection;
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initReplicationSlot(PostgresReplicationConnection.java:86)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.<init>(PostgresReplicationConnection.java:69)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.<init>(PostgresReplicationConnection.java:34)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$ReplicationConnectionBuilder.build(PostgresReplicationConnection.java:284)
    at io.debezium.connector.postgresql.PostgresTaskContext.createReplicationConnection(PostgresTaskContext.java:77)
    at io.debezium.connector.postgresql.RecordsStreamProducer.<init>(RecordsStreamProducer.java:66)
    at io.debezium.connector.postgresql.RecordsSnapshotProducer.<init>(RecordsSnapshotProducer.java:63)
    at io.debezium.connector.postgresql.PostgresConnectorTask.createSnapshotProducer(PostgresConnectorTask.java:148)
    at io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:91)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:141)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:139)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:182)
    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:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)

This is the procedure I've followed:

1) ./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties &
2) ./bin/kafka-server-start ./etc/kafka/server.properties &
3) ./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties &
4) JAR's copied to ./share/java/kafka
5) ./bin/connect-standalone ./etc/schema-registry/connect-avro-standalone.properties ./etc/kafka/test-connector.properties

My properties file has the following contents:

name=test-connector
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=localhost
database.port=5433
database.user=postgres
database.password=postgres
database.dbname=test
database.server.name=test

The connection happens successfully and that can be seen by the log:

[2017-05-11 09:30:28,535] INFO Kafka version : 0.10.2.1-cp1 (org.apache.kafka.common.utils.AppInfoParser:83)
[2017-05-11 09:30:28,535] INFO Kafka commitId : 078e7dc02a100018 (org.apache.kafka.common.utils.AppInfoParser:84)
[2017-05-11 09:30:28,567] INFO Created connector test-connector (org.apache.kafka.connect.cli.ConnectStandalone:90)
[2017-05-11 09:30:28,720] INFO user 'postgres' connected to database 'test' on PostgreSQL 9.6.2 on x86_64-pc-linux-gnu, compiled by gcc (Ubuntu 5.3.1-14ubuntu2) 5.3.1 20160413, 64-bit with roles:
    role 'pg_signal_backend' [superuser: false, replication: false, inherit: true, create role: false, create db: false, can log in: false]
    role 'postgres' [superuser: true, replication: true, inherit: true, create role: true, create db: true, can log in: true] (io.debezium.connector.postgresql.PostgresConnectorTask:81)
[2017-05-11 09:30:28,722] INFO No previous offset found (io.debezium.connector.postgresql.PostgresConnectorTask:85)
[2017-05-11 09:30:28,722] INFO Taking a new snapshot of the DB and streaming logical changes once the snapshot is finished... (io.debezium.connector.postgresql.PostgresConnectorTask:147)

So what is happening is that a first snapshot is being tried, but it fails at commit time. I've found that this is probably because of a different version of PostgreSQL driver, but I was unable to find which one. I am using Java 8 Update 131. Does anyone know how can I solve this?

Thanks in advance,
Igor Felix

[Igor Felix]

I got it! Just copied the correct JDBC version to .share/java/kafka-connect-jdbc and removed the one there and then everything worked!
Hope it helps.

[Naci Ishakbeyoglu]

Igor, I am having the same error when I run them in docker images. So it must be because of wrong jdbc version then? Could you point to which version was that? Isn't it the one in 

postgresql-42.0.0.jar?

Thanks

Naci