[jira] [Closed] (FLINK-38691) Support for Transaction Boundary Events in Flink CDC Connector

["Yanquan Lv (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-38691?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Yanquan Lv closed FLINK-38691.
------------------------------
    Fix Version/s: cdc-3.6.0
         Assignee: Tejansh Rana
       Resolution: Fixed

Merged in master via 7f909f458e49c1b0aaff83484fe13af37da4132e

> Support for Transaction Boundary Events in Flink CDC Connector
> --------------------------------------------------------------
>
>                 Key: FLINK-38691
>                 URL: https://issues.apache.org/jira/browse/FLINK-38691
>             Project: Flink
>          Issue Type: New Feature
>          Components: Flink CDC
>            Reporter: Tejansh Rana
>            Assignee: Tejansh Rana
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: cdc-3.6.0
>
>
> Following my discussion with [~leonard] at Flink Forward, I am writing to propose a feature enhancement for the Flink CDC connectors based on Debezium, related to how it handles transaction metadata.
> *Problem Statement:*
> In data streaming pipelines that require transactional guarantees or need to group atomic changes together, it is essential to identify the boundaries of the original database transaction (i.e., the BEGIN and COMMIT or END events). Currently, the Flink CDC connectors appear to skip these transaction lifecycle events - [https://github.com/apache/flink-cdc/blob/23a1c2efb6fa9ce1c9f17b3836f6aaa995bb0660/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java#L77] 
> [https://github.com/apache/flink-cdc/blob/23a1c2efb6fa9ce1c9f17b3836f6aaa995bb0660/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceRecordEmitter.java#L92]
> This omission makes it challenging to reconstruct the original transaction scope. Without explicit transaction markers, downstream Flink jobs cannot easily guarantee atomicity across sinks.
> *Proposed Solution:*
> The underlying CDC mechanism, Debezium, supports emitting transaction boundary events (BEGIN and END/COMMIT) through its configuration.
> We propose enhancing the Flink CDC connectors to expose this transaction metadata to the Flink pipeline. The connector should emit specialised records or metadata fields that indicate the start and end of a transaction as emitted.
> *Logs of current behaviour:*
> MySQL CDC connector
> {{2025-11-04 14:34:14,033 INFO  org.apache.flink.cdc.connectors.mysql.source.reader.MySqlRecordEmitter [] - Meet unknown element SourceRecord\{sourcePartition={server=mysql_binlog_source}, sourceOffset=\{transaction_id=4541146d-b988-11f0-87f6-0242ac140006:18, file=mysql-bin.000003, pos=5850, gtids=4541146d-b988-11f0-87f6-0242ac140006:1-17, server_id=1}} ConnectRecord\{topic='mysql_binlog_source.transaction', kafkaPartition=null, key=Struct{id=4541146d-b988-11f0-87f6-0242ac140006:18}, keySchema=Schema\{io.debezium.connector.common.TransactionMetadataKey:STRUCT}, value=Struct{color:#FF0000}{status=BEGIN,id=4541146d-b988-11f0-87f6-0242ac140006:18}, valueSchema=Schema\{io.debezium.connector.common.TransactionMetadataValue:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}, just skip{color}.}}
> {color:#172b4d}{{PostgreSQL CDC Connector}}{color}
> {color:#172b4d}{{2025-11-18 09:02:25,544 INFO  org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceRecordEmitter [] - Meet unknown element SourceRecord\{sourcePartition={server=postgres_cdc_source}, sourceOffset=\{transaction_id=843, lsn_proc=31198112, lsn=31198568, txId=843, ts_usec=1763456544336451}} ConnectRecord\{topic='postgres_cdc_source.transaction', kafkaPartition=null, key=Struct{id=843}, keySchema={color:#de350b}Schema\{io.debezium.connector.common.TransactionMetadataKey:STRUCT}, value=Struct\{status=BEGIN,id=843}{color}, valueSchema=Schema\{io.debezium.connector.common.TransactionMetadataValue:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)} for splitState = StreamSplitState\{startingOffset=Offset{lsn=LSN{0/1DC0BA0}, txId=840, lastCommitTs=-9223372036854775808}, endingOffset=Offset\{lsn=LSN{FFFFFFFF/FFFFFFFF}, txId=null, lastCommitTs=-9223372036853775810}, split=StreamSplit\{splitId='stream-split', offset=Offset{lsn=LSN{0/1DC0BA0}, txId=840, lastCommitTs=-9223372036854775808}, endOffset=Offset\{lsn=LSN{FFFFFFFF/FFFFFFFF}, txId=null, lastCommitTs=-9223372036853775810}, isSuspended=false, isSnapshotCompleted=true}}, just skip.}}{color}
>  
> {color:#172b4d}Draft PR for MySQL connector with proposed changes - [https://github.com/apache/flink-cdc/pull/4170]{color}
> {color:#172b4d}I would be happy to create a corresponding PR for the base Incremental Source Record Emitter as well should we accept this proposal{color}



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