[jira] [Created] (FLINK-39633) PostgreSQL CDC backfill throws NullPointerException when WAL stream carries records for other captured tables

["Di Wu (Jira)" <ji...@apache.org>]

Di Wu created FLINK-39633:
-----------------------------

             Summary: PostgreSQL CDC backfill throws NullPointerException when WAL stream carries records for other captured tables
                 Key: FLINK-39633
                 URL: https://issues.apache.org/jira/browse/FLINK-39633
             Project: Flink
          Issue Type: Improvement
          Components: Flink CDC
    Affects Versions: cdc-3.6.0
            Reporter: Di Wu


During the snapshot backfill phase of PostgreSQL CDC (incremental snapshot), the WAL

  stream from the logical replication slot may carry change records for captured tables

  other than the one whose chunk is currently being snapshotted.                                                              

  {\{IncrementalSourceScanFetcher#isChangeRecordInChunkRange}} does not filter records by

  tableId before delegating to \{{JdbcSourceFetchTaskContext#isRecordBetween}}, which calls                                    

  {\{getDatabaseSchema().tableFor(record.tableId)}}. If that table's schema is not yet                                         

  present in \{{RelationAwarePostgresSchema}}'s cache, the lookup returns null and                                             

  {\{ChunkUtils.getSplitColumn}} invokes \{{primaryKeyColumns()}} on null, throwing NPE and                                     

  aborting the snapshot split.                                                                                                

                                                                                                                              

  h2. Stack trace                                                                                                             

                  

  {noformat}

  java.lang.NullPointerException: Cannot invoke "io.debezium.relational.Table.primaryKeyColumns()" because "table" is null

      at org.apache.flink.cdc.connectors.postgres.source.utils.ChunkUtils.getSplitColumn(ChunkUtils.java:45)                  

      at org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext.getSplitType(PostgresSourceFetch

  TaskContext.java:291)                                                                                                       

      at org.apache.flink.cdc.connectors.base.source.reader.external.JdbcSourceFetchTaskContext.isRecordBetween(JdbcSourceFetc

  hTaskContext.java:76)                                                                                                       

      at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.isChangeRecordInChunkRange(I

  ncrementalSourceScanFetcher.java:265)                                                                                       

      at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.pollWithBuffer(IncrementalSo

  urceScanFetcher.java:182)                                                                                                   

      at org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher.pollSplitRecords(Incremental

  SourceScanFetcher.java:122)                                                                                                 

  {noformat}      

                                                                                                                              

  h2. Root cause 

 

  The PostgreSQL logical replication slot streams WAL changes for *all* tables in the

  publication, regardless of which snapshot split is currently being processed. During

  backfill (between low and high watermark), \{{pollWithBuffer}} iterates change events and                                    

  calls \{{isChangeRecordInChunkRange}} on each. The current implementation in

  {\{flink-cdc-base}} is:                                                                                                      

                  

  {code:java}                                                                                                                 

  private boolean isChangeRecordInChunkRange(SourceRecord record) {

      if (taskContext.isDataChangeRecord(record)) {                                                                           

          return taskContext.isRecordBetween(

                  record,                                                                                                     

                  currentSnapshotSplit.getSplitStart(),

                  currentSnapshotSplit.getSplitEnd());                                                                        

      }

      return false;                                                                                                           

  }               

  {code}

 

  It does not verify that the record's tableId matches                                                                        

  {\{currentSnapshotSplit.getTableId()}}. When a record for another captured table flows

  through:                                                                                                                    

                  

  * \{{JdbcSourceFetchTaskContext#getSplitKey}} calls                                                                          

    {\{getDatabaseSchema().tableFor(record.tableId)}};

  * \{{RelationAwarePostgresSchema}} loads relations lazily from \{{Relation}} messages, so                                     

    for a table whose schema has not yet been observed the lookup returns null;                                               

  * \{{ChunkUtils.getSplitColumn(null, ...)}} calls \{{null.primaryKeyColumns()}} -> NPE.                                       

                                                                                                                              

  Even when the schema happens to be cached, comparing a record from table B against chunk                                    

  bounds that were derived from table A's primary key is semantically incorrect and would                                     

  silently pollute the output buffer.                                                                                         

                  

  For contrast, the streaming reader                                                                                          

  (\{{IncrementalSourceStreamFetcher#shouldEmit}}) already defends against this case via

  {\{finishedSplitsInfo.containsKey(tableId)}} before invoking \{{isRecordBetween}}. The scan                                   

  fetcher is missing the symmetric tableId guard.                                                                             

  

  h2. Reproduction                                                                                                            

                  

  * PostgreSQL source with incremental snapshot enabled (default) and

    {\{scan.incremental.snapshot.backfill.skip = false}}.

  * Two or more tables included in the publication, e.g. \{{public.table_a}} and                                               

    {\{public.table_b}}.

  * While Flink CDC is snapshotting \{{table_a}}, generate INSERT/UPDATE traffic on                                            

    {\{table_b}} so its WAL records arrive within \{{table_a}}'s backfill window.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)