Postgres pipeline: NullPointerException in PostgresScanFetchTask

[meng-miao-wu]

Environment

• Flink CDC: 3.5.0
• Flink Version: 1.20

Pipeline Configuration

  source:
    type: postgres
    hostname: '<hostname>'
    port: '<port>'
    username: '<username>'
    password: '<password>'
    tables: 'mydb.public.mytable'
    slot.name: '<slot>'
    debezium.database.dbname: 'mydb'
    decoding.plugin.name: pgoutput
    debezium.publication.name: cdc_publication
    debezium.publication.autocreate.mode: disabled
    scan.startup.mode: initial
    debezium.snapshot.mode: never
    debezium.decimal.handling.mode: double

  sink:
    type: paimon
    catalog.properties.metastore: filesystem
    catalog.properties.warehouse: <warehouse>

  pipeline:
    name: My CDC Pipeline
    parallelism: 1

Error

Caused by: org.apache.flink.util.FlinkRuntimeException: Read split SnapshotSplit{tableId=mydb.public.mytable, splitId='mydb.public.mytable:0', splitKeyType=[`id` BIGINT NOT NULL], splitStart=null, splitEnd=null, highWatermark=null} error due to java.lang.NullPointerException.
	at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.checkReadException(IncrementalSourceScanFetcher.java:205) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.pollSplitRecords(IncrementalSourceScanFetcher.java:116) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.pollSplitRecords(IncrementalSourceSplitReader.java:167) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceSplitReader.fetch(IncrementalSourceSplitReader.java:98) ~[?:?]
	at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58) ~[flink-connector-files-1.20.1.jar:1.20.1]
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) ~[flink-connector-files-1.20.1.jar:1.20.1]
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:117) ~[flink-connector-files-1.20.1.jar:1.20.1]
	at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) ~[?:?]
	at java.util.concurrent.FutureTask.run(Unknown Source) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
	... 1 more
Caused by: io.debezium.DebeziumException: java.lang.NullPointerException
	at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:85) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask.executeDataSnapshot(PostgresScanFetchTask.java:112) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask.execute(AbstractScanFetchTask.java:71) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask.execute(PostgresScanFetchTask.java:86) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.lambda$submitTask$1(IncrementalSourceScanFetcher.java:100) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
	... 1 more
Caused by: java.lang.NullPointerException
	at java.util.Objects.requireNonNull(Unknown Source) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask$PostgresSnapshotSplitReadTask.createDataEvents(PostgresScanFetchTask.java:268) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask$PostgresSnapshotSplitReadTask.doExecute(PostgresScanFetchTask.java:255) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask$PostgresSnapshotSplitReadTask.doExecute(PostgresScanFetchTask.java:211) ~[?:?]
	at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:76) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask.executeDataSnapshot(PostgresScanFetchTask.java:112) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask.execute(AbstractScanFetchTask.java:71) ~[?:?]
	at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresScanFetchTask.execute(PostgresScanFetchTask.java:86) ~[?:?]
	at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.lambda$submitTask$1(IncrementalSourceScanFetcher.java:100) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
	... 1 more

I noticed that SchemaCoordinator registers the table schema without database prefix:

2026-01-16 21:28:02,858 INFO  org.apache.flink.cdc.runtime.operators.schema.regular.SchemaCoordinator [] - Successfully applied schema change event CreateTableEvent{tableId=public.mytable, schema=columns={...} to external system.

Would appreciate guidance:
Am I on the right path? Is the NPE thrown by the mismatch of the tableId?
If yes, did I missed an obvious setup? or a bypass? 😕
If not, what's the issue?

[meng-miao-wu]

I had ran into the same issue before with the postgres connector here
And was able to resolved by adding regex to table-name and state the tables in debezium.table.include.list.
^ no luck resolving with pipeline tho

[meng-miao-wu]

I've identified the root cause of this NullPointerException. The issue stems from a PostgreSQL JDBC driver version mismatch between Flink CDC 3.5.0 and our application:

• Flink CDC 3.5.0 uses PostgreSQL JDBC driver 42.7.3
• Our application was using 42.7.7

Starting in version 42.7.5, the PostgreSQL JDBC driver introduced a fix(pgjdbc/pgjdbc#3390) that changed the behavior of DatabaseMetaData.getTables():

• Before 42.7.5 (including 42.7.3): The TABLE_CAT column returned NULL
• After 42.7.5 (including 42.7.7): The TABLE_CAT column returns the actual database name (via current_database()) to comply with JDBC specification

This behavioral change creates a TableId mismatch:

• When Flink CDC calls readTableNames(), it constructs TableId objects as: new TableId(catalogName, schemaName, tableName) where catalogName = rs.getString(1) (the TABLE_CAT column)
• With 42.7.3: TableId becomes (null, "public", "mytable")
• With 42.7.7: TableId becomes ("mydb", "public", "mytable")
• However, PostgresPipelineRecordEmitter.generateCreateTableEvent() explicitly uses only tableId.schema() and tableId.table(), creating a discrepancy when looking up schema metadata

I was able to align the behavior by downgrade the PostgreSQL JDBC driver to 42.7.3.