clean up

Author: tejanshrana

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

@@ -93,7 +93,6 @@ public MySqlPipelineRecordEmitter(
                 debeziumDeserializationSchema,
                 sourceReaderMetrics,
                 sourceConfig.isIncludeSchemaChanges(),
-                false, // Explicitly disable transaction metadata events
                 false); // Explicitly disable heartbeat events
         this.debeziumDeserializationSchema = debeziumDeserializationSchema;
         this.sourceConfig = sourceConfig;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java

@@ -17,7 +17,6 @@
 
 package org.apache.flink.cdc.connectors.mysql.debezium;
 
-import com.github.shyiko.mysql.binlog.network.SSLMode;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
 import org.apache.flink.cdc.connectors.mysql.source.connection.JdbcConnectionFactory;
 import org.apache.flink.cdc.connectors.mysql.source.offset.BinlogOffset;
@@ -243,26 +242,12 @@ private static Map<String, String> querySystemVariables(
         return variables;
     }
 
-    static SSLMode sslModeFor(MySqlConnectorConfig.SecureConnectionMode mode) {
-        try {
-            return mode == null ? null : SSLMode.valueOf(mode.name());
-        } catch (IllegalArgumentException e) {
-            throw new FlinkRuntimeException(
-              String.format("Invalid SecureConnectionMode provided: %s ", mode.name()), e);
-        }
-    }
-
     public static BinlogOffset findBinlogOffset(
             long targetMs, MySqlConnection connection, MySqlSourceConfig mySqlSourceConfig) {
         MySqlConnection.MySqlConnectionConfiguration config = connection.connectionConfig();
         BinaryLogClient client =
                 new BinaryLogClient(
                         config.hostname(), config.port(), config.username(), config.password());
-        SSLMode sslMode = sslModeFor(config.sslMode());
-        if (sslMode != null) {
-            client.setSSLMode(sslMode);
-        }
-
         if (mySqlSourceConfig.getServerIdRange() != null) {
             client.setServerId(mySqlSourceConfig.getServerIdRange().getStartServerId());
         }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSource.java

@@ -133,7 +133,6 @@ public static <T> MySqlSourceBuilder<T> builder() {
                                 deserializationSchema,
                                 sourceReaderMetrics,
                                 sourceConfig.isIncludeSchemaChanges(),
-                                sourceConfig.isIncludeTransactionMetadataEvents(),
                                 sourceConfig.isIncludeHeartbeatEvents()));
     }
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceBuilder.java

@@ -198,12 +198,6 @@ public MySqlSourceBuilder<T> includeSchemaChanges(boolean includeSchemaChanges)
         return this;
     }
 
-    /** Whether the {@link MySqlSource} should output the transaction metadata events or not. */
-    public MySqlSourceBuilder<T> includeTransactionMetadataEvents(boolean includeTransactionMetadataEvents) {
-        this.configFactory.includeTransactionMetadataEvents(includeTransactionMetadataEvents);
-        return this;
-    }
-
     /** Whether the {@link MySqlSource} should scan the newly added tables or not. */
     public MySqlSourceBuilder<T> scanNewlyAddedTableEnabled(boolean scanNewlyAddedTableEnabled) {
         this.configFactory.scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled);
@@ -249,7 +243,8 @@ public MySqlSourceBuilder<T> heartbeatInterval(Duration heartbeatInterval) {
      * @param heartbeatInterval the interval of heartbeat event
      * @param includeHeartbeatEvents whether to emit heartbeat events
      */
-    public MySqlSourceBuilder<T> heartbeatInterval(Duration heartbeatInterval, boolean includeHeartbeatEvents) {
+    public MySqlSourceBuilder<T> heartbeatInterval(
+            Duration heartbeatInterval, boolean includeHeartbeatEvents) {
         this.configFactory.heartbeatInterval(heartbeatInterval);
         this.configFactory.includeHeartbeatEvents(includeHeartbeatEvents);
         return this;

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java

@@ -62,7 +62,6 @@ public class MySqlSourceConfig implements Serializable {
     private final double distributionFactorUpper;
     private final double distributionFactorLower;
     private final boolean includeSchemaChanges;
-    private final boolean includeTransactionMetadataEvents;
     private final boolean includeHeartbeatEvents;
     private final boolean scanNewlyAddedTableEnabled;
     private final boolean closeIdleReaders;
@@ -101,7 +100,6 @@ public class MySqlSourceConfig implements Serializable {
             double distributionFactorUpper,
             double distributionFactorLower,
             boolean includeSchemaChanges,
-            boolean includeTransactionMetadataEvents,
             boolean includeHeartbeatEvents,
             boolean scanNewlyAddedTableEnabled,
             boolean closeIdleReaders,
@@ -132,7 +130,6 @@ public class MySqlSourceConfig implements Serializable {
         this.distributionFactorUpper = distributionFactorUpper;
         this.distributionFactorLower = distributionFactorLower;
         this.includeSchemaChanges = includeSchemaChanges;
-        this.includeTransactionMetadataEvents = includeTransactionMetadataEvents;
         this.includeHeartbeatEvents = includeHeartbeatEvents;
         this.scanNewlyAddedTableEnabled = scanNewlyAddedTableEnabled;
         this.closeIdleReaders = closeIdleReaders;
@@ -233,10 +230,6 @@ public boolean isIncludeSchemaChanges() {
         return includeSchemaChanges;
     }
 
-    public boolean isIncludeTransactionMetadataEvents() {
-        return includeTransactionMetadataEvents;
-    }
-
     public boolean isIncludeHeartbeatEvents() {
         return includeHeartbeatEvents;
     }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigFactory.java

@@ -63,7 +63,6 @@ public class MySqlSourceConfigFactory implements Serializable {
     private double distributionFactorLower =
             MySqlSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.defaultValue();
     private boolean includeSchemaChanges = false;
-    private boolean includeTransactionMetadataEvents = false;
     private boolean includeHeartbeatEvents = false;
     private boolean scanNewlyAddedTableEnabled = false;
     private boolean closeIdleReaders = false;
@@ -237,12 +236,6 @@ public MySqlSourceConfigFactory includeSchemaChanges(boolean includeSchemaChange
         return this;
     }
 
-    /** Whether the {@link MySqlSource} should output the transaction metadata events or not. */
-    public MySqlSourceConfigFactory includeTransactionMetadataEvents(boolean includeTransactionMetadataEvents) {
-        this.includeTransactionMetadataEvents = includeTransactionMetadataEvents;
-        return this;
-    }
-
     /** Whether the {@link MySqlSource} should output the heartbeat events or not. */
     public MySqlSourceConfigFactory includeHeartbeatEvents(boolean includeHeartbeatEvents) {
         this.includeHeartbeatEvents = includeHeartbeatEvents;
@@ -373,8 +366,6 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
         // Note: the includeSchemaChanges parameter is used to control emitting the schema record,
         // only DataStream API program need to emit the schema record, the Table API need not
         props.setProperty("include.schema.changes", String.valueOf(true));
-        // enable transaction metadata if includeTransactionMetadataEvents is true
-        props.setProperty("provide.transaction.metadata", String.valueOf(includeTransactionMetadataEvents));
         // disable the offset flush totally
         props.setProperty("offset.flush.interval.ms", String.valueOf(Long.MAX_VALUE));
         // disable tombstones
@@ -428,7 +419,6 @@ public MySqlSourceConfig createConfig(int subtaskId, String serverName) {
                 distributionFactorUpper,
                 distributionFactorLower,
                 includeSchemaChanges,
-                includeTransactionMetadataEvents,
                 includeHeartbeatEvents,
                 scanNewlyAddedTableEnabled,
                 closeIdleReaders,

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java

@@ -53,20 +53,17 @@ public class MySqlRecordEmitter<T> implements RecordEmitter<SourceRecords, T, My
     private final DebeziumDeserializationSchema<T> debeziumDeserializationSchema;
     private final MySqlSourceReaderMetrics sourceReaderMetrics;
     private final boolean includeSchemaChanges;
-    private final boolean includeTransactionMetadataEvents;
     private final boolean includeHeartbeatEvents;
     private final OutputCollector<T> outputCollector;
 
     public MySqlRecordEmitter(
             DebeziumDeserializationSchema<T> debeziumDeserializationSchema,
             MySqlSourceReaderMetrics sourceReaderMetrics,
             boolean includeSchemaChanges,
-            boolean includeTransactionMetadataEvents,
             boolean includeHeartbeatEvents) {
         this.debeziumDeserializationSchema = debeziumDeserializationSchema;
         this.sourceReaderMetrics = sourceReaderMetrics;
         this.includeSchemaChanges = includeSchemaChanges;
-        this.includeTransactionMetadataEvents = includeTransactionMetadataEvents;
         this.includeHeartbeatEvents = includeHeartbeatEvents;
         this.outputCollector = new OutputCollector<>();
     }
@@ -111,11 +108,6 @@ protected void processElement(
             if (includeHeartbeatEvents) {
                 emitElement(element, output);
             }
-        } else if (RecordUtils.isTransactionMetadataEvent(element)) {
-            updateStartingOffsetForSplit(splitState, element);
-            if (includeTransactionMetadataEvents) {
-                emitElement(element, output);
-            }
         } else {
             // unknown element
             LOG.info("Meet unknown element {}, just skip.", element);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/RecordUtils.java

@@ -73,8 +73,6 @@ private RecordUtils() {}
             "io.debezium.connector.mysql.SchemaChangeKey";
     public static final String SCHEMA_HEARTBEAT_EVENT_KEY_NAME =
             "io.debezium.connector.common.Heartbeat";
-    public static final String SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME =
-            "io.debezium.connector.common.TransactionMetadataKey";
     private static final DocumentReader DOCUMENT_READER = DocumentReader.defaultReader();
 
     /** Converts a {@link ResultSet} row to an array of Objects. */
@@ -341,18 +339,6 @@ public static boolean isHeartbeatEvent(SourceRecord record) {
                 && SCHEMA_HEARTBEAT_EVENT_KEY_NAME.equalsIgnoreCase(valueSchema.name());
     }
 
-    /**
-     * Check whether the given source record is a transaction metadata event (BEGIN or END).
-     *
-     * <p>Transaction events are emitted by Debezium to mark transaction boundaries when
-     * provide.transaction.metadata is enabled.
-     */
-    public static boolean isTransactionMetadataEvent(SourceRecord record) {
-       Schema keySchema = record.keySchema();
-        return keySchema != null
-                 && SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME.equalsIgnoreCase(keySchema.name());
-    }
-
     /**
      * Return the finished snapshot split information.
      *

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java

@@ -17,9 +17,6 @@
 
 package org.apache.flink.cdc.connectors.mysql.debezium;
 
-import com.github.shyiko.mysql.binlog.network.SSLMode;
-import io.debezium.connector.mysql.MySqlConnectorConfig;
-import java.util.stream.Stream;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfig;
 import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
 import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
@@ -31,9 +28,6 @@
 import java.time.Duration;
 import java.time.ZoneId;
 import java.util.Properties;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
 
 /** Tests for {@link org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils}. */
 class DebeziumUtilsTest {
@@ -95,24 +89,4 @@ private void assertJdbcUrl(String expected, String actual) {
         String[] actualParam = actual.split("&");
         Assertions.assertThat(actualParam).containsExactlyInAnyOrder(expectedParam);
     }
-
-    @ParameterizedTest
-    @MethodSource("sslModeProvider")
-    void testSslModeConversion(MySqlConnectorConfig.SecureConnectionMode input, SSLMode expected) {
-        SSLMode actual = DebeziumUtils.sslModeFor(input);
-        Assertions.assertThat(actual).isEqualTo(expected);
-    }
-    static Stream<Arguments> sslModeProvider() {
-        return Stream.of(
-          Arguments.of(MySqlConnectorConfig.SecureConnectionMode.DISABLED, SSLMode.DISABLED),
-          Arguments.of(
-            MySqlConnectorConfig.SecureConnectionMode.PREFERRED, SSLMode.PREFERRED),
-          Arguments.of(MySqlConnectorConfig.SecureConnectionMode.REQUIRED, SSLMode.REQUIRED),
-          Arguments.of(
-            MySqlConnectorConfig.SecureConnectionMode.VERIFY_CA, SSLMode.VERIFY_CA),
-          Arguments.of(
-            MySqlConnectorConfig.SecureConnectionMode.VERIFY_IDENTITY,
-            SSLMode.VERIFY_IDENTITY));
-    }
-
 }