[Fix][MySQL-CDC]fix recovery task failure caused by binlog deletion (#…

…8587)
apache · Feb 8, 2025 · 087087e · 087087e
1 parent 0f78242
commit 087087e
Show file tree

Hide file tree

Showing 3 changed files with 388 additions and 1 deletion.
diff --git a/...rg/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceRecordEmitter.java b/...rg/apache/seatunnel/connectors/cdc/base/source/reader/IncrementalSourceRecordEmitter.java
@@ -50,6 +50,7 @@
 import static org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.getFetchTimestamp;
 import static org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.getMessageTimestamp;
 import static org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.isDataChangeRecord;
+import static org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.isHeartbeatRecord;
 import static org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils.isSchemaChangeEvent;
 
 /**
@@ -144,7 +145,7 @@ protected void processElement(
             Offset position = getOffsetPosition(element);
             splitState.asIncrementalSplitState().setStartupOffset(position);
             emitElement(element, output);
-        } else if (isDataChangeRecord(element)) {
+        } else if (isDataChangeRecord(element) || isHeartbeatRecord(element)) {
             if (splitState.isIncrementalSplitState()) {
                 Offset position = getOffsetPosition(element);
                 splitState.asIncrementalSplitState().setStartupOffset(position);