[mysql] Original implementation of mysql heartbeat feature

docs/content/connectors/mysql-cdc.md

@@ -375,6 +375,14 @@ After the server you monitored fails in MySQL cluster, you only need to change t
 
 It's recommended to configure a DNS(Domain Name Service) or VIP(Virtual IP Address) for your MySQL cluster, using the DNS or VIP address for ```mysql-cdc``` connector, the DNS or VIP would automatically route the network request to the active MySQL server. In this way, you don't need to modify the address and restart your pipeline anymore.
 
+#### MySQL Heartbeat Event Support
+
+If the table updates infrequently, the binlog file or GTID set may have been cleaned in its last committed binlog position.
+The CDC job restart fails in this case. So the heartbeat event will update binlog position. It's recommended to configure `debezium.heartbeat.interval.ms`.
+```
+debezium.heartbeat.interval.ms=3000
+```
+
 #### How Incremental Snapshot Reading works
 
 When the MySQL CDC source is started, it reads snapshot of table parallelly and then reads binlog of table with single parallelism.

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/MySqlSource.java

@@ -159,7 +159,8 @@ public class MySqlSource<T>
                 new MySqlRecordEmitter<>(
                         deserializationSchema,
                         sourceReaderMetrics,
-                        sourceConfig.isIncludeSchemaChanges()),
+                        sourceConfig.isIncludeSchemaChanges(),
+                        sourceConfig.isHeartbeatEvent()),
                 readerContext.getConfiguration(),
                 mySqlSourceReaderContext,
                 sourceConfig);

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/config/MySqlSourceConfig.java

@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.List;
+import java.util.Optional;
 import java.util.Properties;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -201,4 +202,8 @@ public class MySqlSourceConfig implements Serializable {
     public RelationalTableFilters getTableFilters() {
         return dbzMySqlConfig.getTableFilters();
     }
+
+    public boolean isHeartbeatEvent() {
+        return Optional.ofNullable(dbzConfiguration.getString("heartbeat.interval.ms")).isPresent();
+    }
 }

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java

@@ -40,6 +40,7 @@ import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getHis
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getMessageTimestamp;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.getWatermark;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isDataChangeRecord;
+import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isHeartbeatEvent;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isHighWatermarkEvent;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isSchemaChangeEvent;
 import static com.ververica.cdc.connectors.mysql.source.utils.RecordUtils.isWatermarkEvent;
@@ -61,14 +62,17 @@ public final class MySqlRecordEmitter<T>
     private final MySqlSourceReaderMetrics sourceReaderMetrics;
     private final boolean includeSchemaChanges;
     private final OutputCollector<T> outputCollector;
+    private final boolean heartbeatEvent;
 
     public MySqlRecordEmitter(
             DebeziumDeserializationSchema<T> debeziumDeserializationSchema,
             MySqlSourceReaderMetrics sourceReaderMetrics,
-            boolean includeSchemaChanges) {
+            boolean includeSchemaChanges,
+            boolean heartbeatEvent) {
         this.debeziumDeserializationSchema = debeziumDeserializationSchema;
         this.sourceReaderMetrics = sourceReaderMetrics;
         this.includeSchemaChanges = includeSchemaChanges;
+        this.heartbeatEvent = heartbeatEvent;
         this.outputCollector = new OutputCollector<>();
     }
 
@@ -100,6 +104,11 @@ public final class MySqlRecordEmitter<T>
             }
             reportMetrics(element);
             emitElement(element, output);
+        } else if (heartbeatEvent && isHeartbeatEvent(element)) {
+            if (splitState.isBinlogSplitState()) {
+                BinlogOffset position = getBinlogPosition(element);
+                splitState.asBinlogSplitState().setStartingOffset(position);
+            }
         } else {
             // unknown element
             LOG.info("Meet unknown element {}, just skip.", element);

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/utils/RecordUtils.java

@@ -62,6 +62,8 @@ public class RecordUtils {
 
     public static final String SCHEMA_CHANGE_EVENT_KEY_NAME =
             "io.debezium.connector.mysql.SchemaChangeKey";
+    public static final String SCHEMA_HEARTBEAT_EVENT_KEY_NAME =
+            "io.debezium.connector.common.Heartbeat";
     private static final DocumentReader DOCUMENT_READER = DocumentReader.defaultReader();
 
     /** Converts a {@link ResultSet} row to an array of Objects. */
@@ -298,6 +300,15 @@ public class RecordUtils {
         return false;
     }
 
+    public static boolean isHeartbeatEvent(SourceRecord record) {
+        Schema keySchema = record.keySchema();
+        if (keySchema != null
+                && SCHEMA_HEARTBEAT_EVENT_KEY_NAME.equalsIgnoreCase(keySchema.name())) {
+            return true;
+        }
+        return false;
+    }
+
     /**
      * Return the finished snapshot split information.
      *

flink-connector-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java

@@ -258,7 +258,8 @@ public class MySqlSourceReaderTest extends MySqlSourceTestBase {
                 new MySqlRecordEmitter<>(
                         new ForwardDeserializeSchema(),
                         new MySqlSourceReaderMetrics(metricGroup),
-                        configuration.isIncludeSchemaChanges());
+                        configuration.isIncludeSchemaChanges(),
+                        configuration.isHeartbeatEvent());
         final MySqlSourceReaderContext mySqlSourceReaderContext =
                 new MySqlSourceReaderContext(readerContext);
         return new MySqlSourceReader<>(