[cdc-connector][oracle] Support row_id metadata column and fix headers loss in JdbcSourceFetchTaskContext

[fightBoxing]

Purpose

Add support for extracting Oracle ROWID pseudo-column as a metadata column in Oracle CDC. Users can then access it in Flink SQL via:

CREATE TABLE oracle_source (
    ID INT,
    NAME STRING,
    row_id STRING METADATA FROM 'row_id' VIRTUAL   -- new
) WITH (
    'connector' = 'oracle-cdc',
    'scan.incremental.snapshot.enabled' = 'true',
    ...
);

Typical use cases: data lineage, exact-row identification, idempotent downstream writes, systems that rely on ROWID for reconciliation.

What is changed

1. OracleReadableMetaData — new ROW_ID enum

Reads the ROWID value from the SourceRecord headers (key = ROWID).

2. OracleScanFetchTask (snapshot phase)

Oracle's SELECT * does not include ROWID. So:

• Rewrite the split-scan SQL: SELECT * FROM tab → SELECT T0.*, ROWID FROM tab T0 (ROWID placed at the tail keeps physical column positions unchanged).
• Manually construct ColumnArray from the table's own columns to bypass Debezium's ColumnUtils.toArray(rs, table) strict validation, which would otherwise reject the extra ROWID column.
• Extract ROWID via ((OracleResultSet) rs).getROWID(N+1) and inject it into SourceRecord headers by overriding SnapshotChangeRecordEmitter#getEmitConnectHeaders.

Streaming phase already carries ROWID in headers through Debezium's LogMinerChangeRecordEmitter (from V$LOGMNR_CONTENTS.ROW_ID), so no changes are needed there.

3. JdbcSourceFetchTaskContext — headers-loss bug fix ⭐

While implementing the above, we discovered a latent bug affecting all JDBC-based CDC connectors (MySQL, PostgreSQL, SQL Server, Db2, Oracle):

Both rewriteOutputBuffer() (used for PK-update path) and formatMessageTimestamp() (used for snapshot record normalization) were constructing new SourceRecords using the 8-argument constructor, silently discarding the timestamp and headers from the original record.

Any header-based metadata — including the new Oracle row_id, and any future header-based columns that may be added — would be silently dropped by these two rewrite paths.

Fix: switch both call sites to the 10-argument SourceRecord constructor so timestamp and headers are preserved.

Verification

Compilation:

mvn compile -pl flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc -am -DskipTests

✅ BUILD SUCCESS

Runtime (Oracle Database 12c Standard Edition 12.1.0.2.0, back-ported to release-2.4 and verified end-to-end):

+I[2, bbb, AABDuHAAGAAAAF3AAA]
+I[3, ccc, AABDuHAAGAAAAF0AAA]
+I[4, ddd, AABDuHAAGAAAAF0AAB]
-D[2, bbb, AABDuHAAGAAAAF3AAA]

Both snapshot (READ) and streaming (INSERT/UPDATE/DELETE) rows carry a valid ROWID.

Data flow

Snapshot phase:
  OracleScanFetchTask (rs.getROWID)
    → SnapshotChangeRecordEmitter.getEmitConnectHeaders (inject ROWID)
      → BufferingSnapshotChangeRecordReceiver (new SourceRecord with headers)
        → ChangeEventQueue

Streaming phase (already works):
  LogMinerChangeRecordEmitter.getEmitConnectHeaders (inject ROWID)
    → ChangeEventQueue

Reader side (bug fix here):
  IncrementalSourceScanFetcher.pollSplitRecords
    → JdbcSourceFetchTaskContext.formatMessageTimestamp/rewriteOutputBuffer
      (preserve headers)  ⭐
        → OracleReadableMetaData.ROW_ID.read(record.headers())

User requirements

• scan.incremental.snapshot.enabled=true (this PR does not touch Debezium's native snapshot path).
• Table supplemental logging enabled for streaming capture:

  ALTER TABLE <schema>.<table> ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
  

Compatibility

• No breaking API changes.
• No new configuration options.
• Users who don't declare row_id metadata column see zero behavioral change (SQL rewrite still runs but only adds a single pseudo-column to the JDBC query, which is negligible).
• The JdbcSourceFetchTaskContext fix is transparent — records that previously had null/absent headers still have null/absent headers afterwards.