CDC with WSO2 Streaming Integrator and Postgres DB

Question

I am trying to setup Change Data Capture (CDC) between WSO2 Streaming Integrator and a local Postgres DB.

I have added the Postgres Driver (v42.2.5) to SI_HOME/lib and I am able to read data from the database from a Siddhi application.

I am following the CDCWithListeningMode example to implement CDC and I am using pgoutput as the logical decoding plugin. But when I run the application I get the following log.

    [2020-04-23_19-02-37_460] INFO {org.apache.kafka.connect.json.JsonConverterConfig} - JsonConverterConfig values: 
    converter.type = key
    schemas.cache.size = 1000
    schemas.enable = true

[2020-04-23_19-02-37_461] INFO {org.apache.kafka.connect.json.JsonConverterConfig} - JsonConverterConfig values: 
    converter.type = value
    schemas.cache.size = 1000
    schemas.enable = false

[2020-04-23_19-02-37_461] INFO {io.debezium.embedded.EmbeddedEngine$EmbeddedConfig} - EmbeddedConfig values: 
    access.control.allow.methods = 
    access.control.allow.origin = 
    bootstrap.servers = [localhost:9092]
    header.converter = class org.apache.kafka.connect.storage.SimpleHeaderConverter
    internal.key.converter = class org.apache.kafka.connect.json.JsonConverter
    internal.value.converter = class org.apache.kafka.connect.json.JsonConverter
    key.converter = class org.apache.kafka.connect.json.JsonConverter
    listeners = null
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    offset.flush.interval.ms = 60000
    offset.flush.timeout.ms = 5000
    offset.storage.file.filename = 
    offset.storage.partitions = null
    offset.storage.replication.factor = null
    offset.storage.topic = 
    plugin.path = null
    rest.advertised.host.name = null
    rest.advertised.listener = null
    rest.advertised.port = null
    rest.host.name = null
    rest.port = 8083
    ssl.client.auth = none
    task.shutdown.graceful.timeout.ms = 5000
    value.converter = class org.apache.kafka.connect.json.JsonConverter

[2020-04-23_19-02-37_516] INFO {io.debezium.connector.common.BaseSourceTask} -    offset.storage = io.siddhi.extension.io.cdc.source.listening.InMemoryOffsetBackingStore 
[2020-04-23_19-02-37_517] INFO {io.debezium.connector.common.BaseSourceTask} -    database.server.name = localhost_5432 
[2020-04-23_19-02-37_517] INFO {io.debezium.connector.common.BaseSourceTask} -    database.port = 5432 
[2020-04-23_19-02-37_517] INFO {io.debezium.connector.common.BaseSourceTask} -    table.whitelist = SweetProductionTable 
[2020-04-23_19-02-37_517] INFO {io.debezium.connector.common.BaseSourceTask} -    cdc.source.object = 1716717434 
[2020-04-23_19-02-37_517] INFO {io.debezium.connector.common.BaseSourceTask} -    database.hostname = localhost 
[2020-04-23_19-02-37_518] INFO {io.debezium.connector.common.BaseSourceTask} -    database.password = ******** 
[2020-04-23_19-02-37_518] INFO {io.debezium.connector.common.BaseSourceTask} -    name = CDCWithListeningModeinsertSweetProductionStream 
[2020-04-23_19-02-37_518] INFO {io.debezium.connector.common.BaseSourceTask} -    server.id = 6140 
[2020-04-23_19-02-37_519] INFO {io.debezium.connector.common.BaseSourceTask} -    database.history = io.debezium.relational.history.FileDatabaseHistory 
[2020-04-23_19-02-38_103] INFO {io.debezium.connector.postgresql.PostgresConnectorTask} - user 'user_name' connected to database 'db_name' on PostgreSQL 11.5, compiled by Visual C++ build 1914, 64-bit with roles:
    role 'user_name' [superuser: false, replication: true, inherit: true, create role: false, create db: false, can log in: true] (Encoded) 
[2020-04-23_19-02-38_104] INFO {io.debezium.connector.postgresql.PostgresConnectorTask} - No previous offset found 
[2020-04-23_19-02-38_104] INFO {io.debezium.connector.postgresql.PostgresConnectorTask} - Taking a new snapshot of the DB and streaming logical changes once the snapshot is finished... 
[2020-04-23_19-02-38_105] INFO {io.debezium.util.Threads} - Requested thread factory for connector PostgresConnector, id = localhost_5432 named = records-snapshot-producer 
[2020-04-23_19-02-38_105] INFO {io.debezium.util.Threads} - Requested thread factory for connector PostgresConnector, id = localhost_5432 named = records-stream-producer 
[2020-04-23_19-02-38_293] INFO {io.debezium.connector.postgresql.connection.PostgresConnection} - Obtained valid replication slot ReplicationSlot [active=false, latestFlushedLSN=null] 
[2020-04-23_19-02-38_704] ERROR {io.siddhi.core.stream.input.source.Source} - Error on 'CDCWithListeningMode'. Connection to the database lost. Error while connecting at Source 'cdc' at 'insertSweetProductionStream'. Will retry in '5 sec'. (Encoded) 
io.siddhi.core.exception.ConnectionUnavailableException: Connection to the database lost.
    at io.siddhi.extension.io.cdc.source.CDCSource.lambda$connect$1(CDCSource.java:424)
    at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:793)
    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.ConnectException: Cannot create replication connection
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.(PostgresReplicationConnection.java:87)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.(PostgresReplicationConnection.java:38)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$ReplicationConnectionBuilder.build(PostgresReplicationConnection.java:362)
    at io.debezium.connector.postgresql.PostgresTaskContext.createReplicationConnection(PostgresTaskContext.java:65)
    at io.debezium.connector.postgresql.RecordsStreamProducer.(RecordsStreamProducer.java:81)
    at io.debezium.connector.postgresql.RecordsSnapshotProducer.(RecordsSnapshotProducer.java:70)
    at io.debezium.connector.postgresql.PostgresConnectorTask.createSnapshotProducer(PostgresConnectorTask.java:133)
    at io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:86)
    at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:45)
    at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:677)
    ... 3 more
Caused by: io.debezium.jdbc.JdbcConnectionException: ERROR: could not access file "decoderbufs": No such file or directory
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initReplicationSlot(PostgresReplicationConnection.java:145)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.(PostgresReplicationConnection.java:79)
    ... 12 more
Caused by: org.postgresql.util.PSQLException: ERROR: could not access file "decoderbufs": No such file or directory
    at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2440)
    at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2183)
    at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:308)
    at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:441)
    at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:365)
    at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:307)
    at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:293)
    at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:270)
    at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:266)
    at org.postgresql.replication.fluent.logical.LogicalCreateSlotBuilder.make(LogicalCreateSlotBuilder.java:48)
    at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initReplicationSlot(PostgresReplicationConnection.java:108)
    ... 13 more

Debezium defaults to decoderbufs plugin - "could not access file "decoderbufs": No such file or directory".

According to this answer, the issue is due to the configuration of decoderbufs plugin.

Details

Postgres - 11.4
siddhi-cdc-io - 2.0.3
Debezium - 0.8.3

How do I configure the embedded debezium engine to use the pgoutput plugin? Will changing this configuration fix the error?

Please help me with this issue. I have not found any resources that can help me.

score 0 · Answer 1 · answered Apr 24 '20 at 13:34

you either need to update the Debezium to the latest 1.1 version - this will enable you to use pgoutput plugin using plugin.name config option or you need to deploy (and maybe build) decoderbufs.so library to your PostgreSQL database.

I'd recommend the former as 0.8.3 is very old version.

score 0 · Answer 2 · answered Aug 26 '20 at 16:33

I observed this behavior with PostgreSQL 12 when I tried to do CDC with pgoutput logical decoding output plug-in. It seems like even though I configured the database with pgoutput, the siddhi extension is trying to make the connection using "decoderbufs" as decoding plug-in.

When I tried configuring decoderbufs as the logical decoding output plug-in in the database level, I was able to use siddhi io extension without any issue.

It seems like for now, Siddhi io CDC only supports decoderbufs logical decoding output plug-in with PostgreSQL.

CDC with WSO2 Streaming Integrator and Postgres DB

2 Answers2