yuxiqian commented on code in PR #3812:
URL: https://github.com/apache/flink-cdc/pull/3812#discussion_r2017811453
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlPipelineRecordEmitter.java:
##########
@@ -90,7 +94,16 @@ public MySqlPipelineRecordEmitter(
protected void processElement(
SourceRecord element, SourceOutput<Event> output, MySqlSplitState
splitState)
throws Exception {
- if (isLowWatermarkEvent(element) && splitState.isSnapshotSplitState())
{
+ if
(StartupOptions.snapshot().equals(sourceConfig.getStartupOptions())) {
+ // In snapshot mode, we simply emit all schemas at once.
+ if (!alreadySendAllCreateTableTables) {
Review Comment:
In snapshot mode, we will:
* obtain and check startup mode
* then check the flag
In any other modes, we will:
* obtain and check startup mode
I would suggest naming `alreadySendAllCreateTableTables` =>
`shouldEmitAllCtesInSnapshotMode`, and set it to `true` in snapshot mode,
`false` (in other modes), and the checking could be simplified to:
```java
if (shouldEmitAllCtesInSnapshotMode) {
createTableEventCache.forEach(
(tableId, createTableEvent) -> output.collect(createTableEvent)
);
shouldEmitAllCtesInSnapshotMode = false;
}
```
so we don't have to check the startup mode every time when we receive a
SourceRecord.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
