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

["Tejansh Rana (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-38691?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18040437#comment-18040437 ] 

Tejansh Rana commented on FLINK-38691:
--------------------------------------

Discussed in email thread:
 # This feature should be supported in the DataStream API since Flink SQL/Table API relies on RowData.
 # To avoid a breaking change and performance overhead of parsing these events, this feature should be supported by an option to enable/disable parsing and emitting of these events. This option should be disabled by default.
 # MySQL pipeline connector is out of scope of this ticket. 

> 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
>            Priority: Major
>
> 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)