[source-postgres] CDC Not Working With Connector Version 3.6.22 (PgJDBC Issue)

[hagar-yasser1]

Connector Name

source-postgres

Connector Version

3.6.22

What step the error happened?

During the sync

Relevant information

Summary

After upgrading the source version to 3.6.22, and on running Postgres Sync Job using CDC, we face the issue of repeatedly failing pipelines or repeatedly refreshing pipelines (depending on the value for Invalid CDC position behavior Property). We traced back the issue to the PgJDBC version used. The version injected into Debezium by Airbyte automatically advances the confirmed_flush_lsn in cases there are no changes for the publication tables in the replication slot.

Details

1. In case Invalid CDC position behavior (Advanced) Property is Fail sync, Output Log is:

   Saved offset is before replication slot's confirmed lsn. Please reset the connection, and then increase WAL retention and/or increase sync frequency to prevent this from happening in the future. See https://docs.airbyte.com/integrations/sources/postgres/postgres-troubleshooting#under-cdc-incremental-mode-there-are-still-full-refresh-syncs for more details.
   

2. In case Invalid CDC position behavior (Advanced) Property is Re-sync data, Output Log is:

   Saved offset is before Replication slot's confirmed_flush_lsn, Airbyte will trigger sync from scratch
   

The cause for this issue is that starting from postgres source 3.6.19 the datastore-postgres started using api 'org.postgresql:postgresql:42.7.4'

• datastore-postgres is taken as a dependency in source-postgres

  airbyteJavaConnector {  cdkVersionRequired = '0.48.2'  features = ['db-sources', 'datastore-postgres']  useLocalCdk = false }
  

• All versions of the PgJDBC starting 42.7.0 cannot be used with debezium because it automatically flushes the LSN in case there is no new events are coming out of the replication stream given a publication but the WAL is still full with changes out of the used publication. Refer to the following links:

  1. Debezium opened issue on PgJDBC here Keep Alive Feature Flushes Replication Slot Unexpectedly pgjdbc/pgjdbc#3138
  2. The version used by Debezium-2.6.0-Final for Postgres here https://github.com/debezium/debezium/blob/4159397bd40dd5d0c8efe2b97ba61ba8eb220b58/pom.xml#L128
  3. The PR on PgJDBC and issue here feat: Use KeepAlive to confirm LSNs pgjdbc/pgjdbc#2941
  4. Zalando's post regarding the PR on PgJDBC here https://engineering.zalando.com/posts/2023/11/patching-pgjdbc.html

• In the related logs, the PgJDBC version used is only visible when there is a transient error that takes place when closing the stream.

Proposed Solutions

• Downgrade the source-postgres to 3.6.18
• Downgrade the PgJDBC version used by datastore-postgres to anything less than 42.7.0

Related Airbyte Issues

• [source-postgres] Change Data Capture (CDC) Not Working #47547

Relevant log output

2024-11-30 00:40:20,219 [pool-10-thread-4]	�[1;31mERROR�[0;39m	i.a.w.i.VersionedAirbyteStreamFactory(internalLog):310 - ERROR debezium-postgresconnector-wms_db-change-event-source-coordinator i.d.c.p.c.PostgresReplicationConnection(close):775 Unexpected error while closing Postgres connection org.postgresql.util.PSQLException: Unable to close connection properly
	at org.postgresql.jdbc.PgConnection.close(PgConnection.java:870) ~[postgresql-42.7.4.jar:42.7.4]
	at io.debezium.jdbc.JdbcConnection.lambda$doClose$4(JdbcConnection.java:951) ~[debezium-core-2.6.2.Final.jar:2.6.2.Final]
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
	at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
Caused by: java.net.SocketException: Broken pipe
	at java.base/sun.nio.ch.SocketDispatcher.write0(Native Method) ~[?:?]
	at java.base/sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:62) ~[?:?]
	at java.base/sun.nio.ch.NioSocketImpl.tryWrite(NioSocketImpl.java:394) ~[?:?]
	at java.base/sun.nio.ch.NioSocketImpl.implWrite(NioSocketImpl.java:410) ~[?:?]
	at java.base/sun.nio.ch.NioSocketImpl.write(NioSocketImpl.java:440) ~[?:?]
	at java.base/sun.nio.ch.NioSocketImpl$2.write(NioSocketImpl.java:819) ~[?:?]
	at java.base/java.net.Socket$SocketOutputStream.write(Socket.java:1195) ~[?:?]
	at java.base/sun.security.ssl.SSLSocketOutputRecord.deliver(SSLSocketOutputRecord.java:345) ~[?:?]
	at java.base/sun.security.ssl.SSLSocketImpl$AppOutputStream.write(SSLSocketImpl.java:1304) ~[?:?]
	at org.postgresql.util.internal.PgBufferedOutputStream.flushBuffer(PgBufferedOutputStream.java:41) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.util.internal.PgBufferedOutputStream.flush(PgBufferedOutputStream.java:48) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.core.PGStream.flush(PGStream.java:707) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.core.QueryExecutorCloseAction.close(QueryExecutorCloseAction.java:73) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.jdbc.PgConnectionCleaningAction.onClean(PgConnectionCleaningAction.java:89) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.util.LazyCleaner$Node.onClean(LazyCleaner.java:219) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.util.LazyCleaner$Node.clean(LazyCleaner.java:210) ~[postgresql-42.7.4.jar:42.7.4]
	at org.postgresql.jdbc.PgConnection.close(PgConnection.java:867) ~[postgresql-42.7.4.jar:42.7.4]
	... 5 more

Contribute

• Yes, I want to contribute

[marcosmarxm]

@airbytehq/dbsources can someone take a look into this issue?

[theyueli]

@yardencarmeli could you take a look?