[source-postgres] CDC sync is making the confirmed flush LSN to a value > saved state LSN

[vrishin-bolt]

Connector Name

source-postgres

Connector Version

v3.6.28

What step the error happened?

During the sync

Relevant information

Hi all,

Something super weird is happening -

1. I am doing PG -> BQ, sync in CDC mode (im in latest PG source and BQ destination version and airbyte version 0.64.2)
2. This is the first sync => its going to load everything

Confirmed flush LSN (2 mins before sync start)
-- airbyte_slot_2 C1/20E83D08 829480779016

Confirmed flush LSN (2 mins after sync start)
-- airbyte_slot_2 C1/22539BC0 829504592832

State LSN saved in airbyte state file
829504592784

⁉️ Clearly state LSN < confirmed flush LSN
Airbyte is saving an LSN > confirmed flush LSN when it starts? (or)
Airbyte is confirm that it has read and processed a LSN > what it has saved? Why??

This is causing my sync to fail with
"Saved offset is before replication slot's confirmed LSN" after sometime

I have attached the logs (I stopped the sync the moment I noticed this happened)

Airbyte Logs.txt

Relevant log output

Contribute

• Yes, I want to contribute

[vrishin-bolt]

Here is an update:
Doing PG -> BQ via CDC, noticing the shared state LSN that is being saved is incorrect(on the very first sync in incremental| append + deduped mode). Here are the relevant log(s) -

2025-03-28 01:11:44 source INFO main i.d.c.CommonConnectorConfig(getSourceInfoStructMaker):1649 Loading the custom source info struct maker plugin: io.debezium.connector.postgresql.PostgresSourceInfoStructMaker
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(extractLsn):191 Found previous partition offset PostgresPartition [sourcePartition={server=api_staging}]: {lsn=833635403600, txId=5209687748, ts_usec=1743104503698506}
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(parseSavedOffset):171 Closing offsetStorageReader and fileOffsetBackingStore
2025-03-28 01:11:44 source INFO main o.a.k.c.s.FileOffsetBackingStore(stop):71 Stopped FileOffsetBackingStore
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(isSavedOffsetAfterReplicationSlotLSN):69 Replication slot confirmed_flush_lsn : 833635145880 Saved offset LSN : 833635403600
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresCdcCtidInitializer(getCtidInitialLoadGlobalStateManager):117 Streams to be synced via ctid (can include RFR streams) : 1
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresCdcCtidInitializer(getCtidInitialLoadGlobalStateManager):118 Streams: public.risk_assessment_factors
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresQueryUtils(fileNodeForIndividualStream):235 Relation filenode is for stream "public"."risk_assessment_factors" is 1270938
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(format):240 Initial Debezium state constructed: {"[\"api_staging\",{\"server\":\"api_staging\"}]":"{\"transaction_id\":null,\"lsn\":833635403792,\"txId\":5209687750,\"ts_usec\":1743104504606614}"}
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresUtils(isCdc):70 using CDC: true
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresSource(getIncrementalIterators):508 Using ctid + CDC
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresUtils(getFirstRecordWaitTime):171 First record waiting time: 120 seconds
2025-03-28 01:11:44 source WARN main i.a.c.i.s.r.InitialLoadTimeoutUtil(getInitialLoadTimeout):36 Initial Load timeout is overridden to 24 hours, which is the max time allowed for safety.
2025-03-28 01:11:44 source INFO main i.a.c.i.s.r.InitialLoadTimeoutUtil(getInitialLoadTimeout):44 Initial Load timeout: 86400 seconds
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresCdcCtidInitializer(cdcCtidIteratorsCombined):149 First record waiting time: 120 seconds
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresCdcCtidInitializer(cdcCtidIteratorsCombined):150 Initial load timeout: 24 hours
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresCdcCtidInitializer(cdcCtidIteratorsCombined):151 Queue size: 10000
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(format):240 Initial Debezium state constructed: {"[\"api_staging\",{\"server\":\"api_staging\"}]":"{\"transaction_id\":null,\"lsn\":833635403840,\"txId\":5209687751,\"ts_usec\":1743104504691364}"}
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresUtils(shouldFlushAfterSync):78 Should flush after sync: true
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresSource(toSslJdbcParamInternal):924 DISABLED toSslJdbcParam disable
2025-03-28 01:11:44 source INFO main o.a.k.c.c.AbstractConfig(logAll):370 JsonConverterConfig values: 
2025-03-28 01:11:44 source INFO main o.a.k.c.c.AbstractConfig(logAll):370 StandaloneConfig values: 
	access.control.allow.methods = 
	access.control.allow.origin = 
	admin.listeners = null
	auto.include.jmx.reporter = true
	bootstrap.servers = [localhost:9092]
	client.dns.lookup = use_all_dns_ips
	config.providers = []
	connector.client.config.override.policy = All
	header.converter = class org.apache.kafka.connect.storage.SimpleHeaderConverter
	key.converter = class org.apache.kafka.connect.json.JsonConverter
	listeners = [http://:8083]
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	offset.flush.interval.ms = 1000
	offset.flush.timeout.ms = 5000
	offset.storage.file.filename = /tmp/cdc-state-offset13844630858619352820/offset.dat
	plugin.discovery = hybrid_warn
	plugin.path = null
	response.http.headers.config = 
	rest.advertised.host.name = null
	rest.advertised.listener = null
	rest.advertised.port = null
	rest.extension.classes = []
	ssl.cipher.suites = null
	ssl.client.auth = none
	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
	ssl.endpoint.identification.algorithm = https
	ssl.engine.factory.class = null
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.certificate.chain = null
	ssl.keystore.key = null
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLSv1.3
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.certificates = null
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	task.shutdown.graceful.timeout.ms = 5000
	topic.creation.enable = true
	topic.tracking.allow.reset = true
	topic.tracking.enable = true
	value.converter = class org.apache.kafka.connect.json.JsonConverter
2025-03-28 01:11:44 source INFO main o.a.k.c.s.FileOffsetBackingStore(start):63 Starting FileOffsetBackingStore with file /tmp/cdc-state-offset13844630858619352820/offset.dat
2025-03-28 01:11:44 source INFO main o.a.k.c.c.AbstractConfig(logAll):370 JsonConverterConfig values: 
	converter.type = key
	decimal.format = BASE64
	replace.null.with.default = true
	schemas.cache.size = 1000
	schemas.enable = false
2025-03-28 01:11:44 source INFO main o.a.k.c.c.AbstractConfig(logAll):370 JsonConverterConfig values: 
	converter.type = value
	decimal.format = BASE64
	replace.null.with.default = true
	schemas.cache.size = 1000
	schemas.enable = false
2025-03-28 01:11:44 source INFO main i.d.c.CommonConnectorConfig(getSourceInfoStructMaker):1649 Loading the custom source info struct maker plugin: io.debezium.connector.postgresql.PostgresSourceInfoStructMaker
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(extractLsn):191 Found previous partition offset PostgresPartition [sourcePartition={server=api_staging}]: {lsn=833635403840, txId=5209687751, ts_usec=1743104504691364}
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(parseSavedOffset):171 Closing offsetStorageReader and fileOffsetBackingStore
2025-03-28 01:11:44 source INFO main o.a.k.c.s.FileOffsetBackingStore(stop):71 Stopped FileOffsetBackingStore
2025-03-28 01:11:44 source INFO main i.a.i.s.p.PostgresUtils(shouldFlushAfterSync):78 Should flush after sync: true
2025-03-28 01:11:44 source INFO main i.a.i.s.p.c.PostgresDebeziumStateUtil(commitLSNToPostgresDatabase):104 Committing upto LSN: 833635403840

I notice from these logs

Initial Debezium state constructed: {"[\"api_staging\",{\"server\":\"api_staging\"}]":"{\"transaction_id\":null,\"lsn\":833635403792,\"txId\":5209687750,\"ts_usec\":1743104504606614}"}

and

 Initial Debezium state constructed: {"[\"api_staging\",{\"server\":\"api_staging\"}]":"{\"transaction_id\":null,\"lsn\":833635403840,\"txId\":5209687751,\"ts_usec\":1743104504691364}"}

• but only the first one(incorrect and old) is saved, and the second one(correct and latest) is not. Why is that?
• I'm on source-postgres v3.6.18 (not on latest cos there is some issue #49802, that says latest one has some cdc issues)
• I'm on latest airbyte v1.5.0
• I have also correctly configured and validated both the debezium heartbeat query and the pg_cron heartbeat
• I'm running on AWS Aurora PG(this is a standalone writer instance that is getting AWS CDC replicated data from another PG writer instance) with rds.logical_wal_cache=0, max_slot_wal_keep_size=-1

edit1: btw I also tried with v3.6.28 (same issue)
edit2: I think im facing this issue

[marcosmarxm]

@rodireich do you mind taking a look at this issue?

[killix]

I encountered the same issue while testing with a fresh Airbyte installation, using the Postgres connector v3.6.30 and Airbyte v1.5.

The issue occurred in my test environment, which has minimal traffic and only one table involved. While the workaround of switching the CDC mode from Fail sync to Re-sync data works in this case, it’s not viable for our production environment due to the presence of tables with millions of rows.

Updating the connection state using the confirmed_flush_lsn value from the following query:

SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = 'airbyte_slot';

will make the next sync working, but not the next after (back to the initial issue)

If someone from the team is looking into this, please note that this issue currently blocks us from upgrading to Airbyte v1.5 in production.