add close FetchTask.Context

apache · Mar 13, 2023 · fc918b9 · fc918b9
1 parent ef4384b
commit fc918b9
Show file tree

Hide file tree

Showing 8 changed files with 53 additions and 29 deletions.
diff --git a/.../main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/FetchTask.java b/.../main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/FetchTask.java
@@ -66,5 +66,7 @@ interface Context {
         void rewriteOutputBuffer(Map<Struct, SourceRecord> outputBuffer, SourceRecord changeRecord);
 
         List<SourceRecord> formatMessageTimestamp(Collection<SourceRecord> snapshotRecords);
+
+        void close();
     }
 }
diff --git a/...he/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java b/...he/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java
@@ -192,6 +192,9 @@ private void checkReadException() {
     @Override
     public void close() {
         try {
+            if (taskContext != null) {
+                taskContext.close();
+            }
             if (snapshotSplitReadTask != null) {
                 snapshotSplitReadTask.shutdown();
             }

diff --git a/.../seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java b/.../seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
@@ -121,6 +121,9 @@ private void checkReadException() {
     @Override
     public void close() {
         try {
+            if (taskContext != null) {
+                taskContext.close();
+            }
             if (streamFetchTask != null) {
                 streamFetchTask.shutdown();
             }

diff --git a/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlDialect.java b/...rc/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlDialect.java
@@ -33,17 +33,13 @@
 import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlSchema;
 import org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.TableDiscoveryUtils;
 
-import com.github.shyiko.mysql.binlog.BinaryLogClient;
-import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
 
 import java.sql.SQLException;
 import java.util.List;
 
-import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.createBinaryClient;
-import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.createMySqlConnection;
 import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.isTableIdCaseSensitive;
 
 /** The {@link JdbcDataSourceDialect} implementation for MySQL datasource. */
@@ -104,12 +100,7 @@ public TableChanges.TableChange queryTableSchema(JdbcConnection jdbc, TableId ta
     @Override
     public MySqlSourceFetchTaskContext createFetchTaskContext(
             SourceSplitBase sourceSplitBase, JdbcSourceConfig taskSourceConfig) {
-        final MySqlConnection jdbcConnection =
-                createMySqlConnection(taskSourceConfig.getDbzConfiguration());
-        final BinaryLogClient binaryLogClient =
-                createBinaryClient(taskSourceConfig.getDbzConfiguration());
-        return new MySqlSourceFetchTaskContext(
-                taskSourceConfig, this, jdbcConnection, binaryLogClient);
+        return new MySqlSourceFetchTaskContext(taskSourceConfig, this);
     }
 
     @Override

diff --git a/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java b/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java
@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.reader.fetch;
 
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.ExceptionUtils;
 import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
 import org.apache.seatunnel.connectors.cdc.base.dialect.JdbcDataSourceDialect;
 import org.apache.seatunnel.connectors.cdc.base.relational.JdbcSourceEventDispatcher;
@@ -62,15 +63,21 @@
 import io.debezium.schema.DataCollectionId;
 import io.debezium.schema.TopicSelector;
 import io.debezium.util.Collect;
+import lombok.extern.slf4j.Slf4j;
 
+import java.io.IOException;
+import java.sql.SQLException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
 import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source.offset.BinlogOffset.BINLOG_FILENAME_OFFSET_KEY;
+import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.createBinaryClient;
+import static org.apache.seatunnel.connectors.seatunnel.cdc.mysql.utils.MySqlConnectionUtils.createMySqlConnection;
 
 /** The context for fetch task that fetching data of snapshot split from MySQL data source. */
+@Slf4j
 public class MySqlSourceFetchTaskContext extends JdbcSourceFetchTaskContext {
 
     private static final Logger LOG = LoggerFactory.getLogger(MySqlSourceFetchTaskContext.class);
@@ -89,13 +96,10 @@ public class MySqlSourceFetchTaskContext extends JdbcSourceFetchTaskContext {
     private MySqlErrorHandler errorHandler;
 
     public MySqlSourceFetchTaskContext(
-            JdbcSourceConfig sourceConfig,
-            JdbcDataSourceDialect dataSourceDialect,
-            MySqlConnection connection,
-            BinaryLogClient binaryLogClient) {
+            JdbcSourceConfig sourceConfig, JdbcDataSourceDialect dataSourceDialect) {
         super(sourceConfig, dataSourceDialect);
-        this.connection = connection;
-        this.binaryLogClient = binaryLogClient;
+        this.connection = createMySqlConnection(sourceConfig.getDbzConfiguration());
+        this.binaryLogClient = createBinaryClient(sourceConfig.getDbzConfiguration());
         this.metadataProvider = new MySqlEventMetadataProvider();
     }
 
@@ -159,6 +163,18 @@ public void configure(SourceSplitBase sourceSplitBase) {
         this.errorHandler = new MySqlErrorHandler(connectorConfig.getLogicalName(), queue);
     }
 
+    @Override
+    public void close() {
+        try {
+            this.connection.close();
+            this.binaryLogClient.disconnect();
+        } catch (SQLException e) {
+            log.warn("Failed to close connection, {}", ExceptionUtils.getMessage(e));
+        } catch (IOException e) {
+            log.warn("Failed to close binaryLogClient, {}", ExceptionUtils.getMessage(e));
+        }
+    }
+
     @Override
     public MySqlSourceConfig getSourceConfig() {
         return (MySqlSourceConfig) sourceConfig;

diff --git a/...l/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java b/...l/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java
@@ -339,6 +339,7 @@ public static String quote(TableId tableId) {
     private static PreparedStatement initStatement(JdbcConnection jdbc, String sql, int fetchSize)
             throws SQLException {
         final Connection connection = jdbc.connection();
+        // Add MySQL metadata locks to prevent modification of table structure.
         connection.setAutoCommit(false);
         final PreparedStatement statement =
                 connection.prepareStatement(

diff --git a/...g/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/SqlServerDialect.java b/...g/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/SqlServerDialect.java
@@ -34,7 +34,6 @@
 import org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.utils.SqlServerSchema;
 import org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.utils.TableDiscoveryUtils;
 
-import io.debezium.connector.sqlserver.SqlServerConnection;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
@@ -104,13 +103,8 @@ public TableChanges.TableChange queryTableSchema(JdbcConnection jdbc, TableId ta
     @Override
     public SqlServerSourceFetchTaskContext createFetchTaskContext(
             SourceSplitBase sourceSplitBase, JdbcSourceConfig taskSourceConfig) {
-        final SqlServerConnection jdbcConnection =
-                createSqlServerConnection(taskSourceConfig.getDbzConfiguration());
-        final SqlServerConnection metaDataConnection =
-                createSqlServerConnection(taskSourceConfig.getDbzConfiguration());
 
-        return new SqlServerSourceFetchTaskContext(
-                taskSourceConfig, this, jdbcConnection, metaDataConnection);
+        return new SqlServerSourceFetchTaskContext(taskSourceConfig, this);
     }
 
     @Override

diff --git a/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/SqlServerSourceFetchTaskContext.java b/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/SqlServerSourceFetchTaskContext.java
@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.source.reader.fetch;
 
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.ExceptionUtils;
 import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
 import org.apache.seatunnel.connectors.cdc.base.dialect.JdbcDataSourceDialect;
 import org.apache.seatunnel.connectors.cdc.base.relational.JdbcSourceEventDispatcher;
@@ -58,13 +59,18 @@
 import io.debezium.schema.DataCollectionId;
 import io.debezium.schema.TopicSelector;
 import io.debezium.util.Collect;
+import lombok.extern.slf4j.Slf4j;
 
+import java.sql.SQLException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.utils.SqlServerConnectionUtils.createSqlServerConnection;
+
 /** The context for fetch task that fetching data of snapshot split from MySQL data source. */
+@Slf4j
 public class SqlServerSourceFetchTaskContext extends JdbcSourceFetchTaskContext {
 
     private final SqlServerConnection dataConnection;
@@ -83,13 +89,11 @@ public class SqlServerSourceFetchTaskContext extends JdbcSourceFetchTaskContext
     private SnapshotChangeEventSourceMetrics snapshotChangeEventSourceMetrics;
 
     public SqlServerSourceFetchTaskContext(
-            JdbcSourceConfig sourceConfig,
-            JdbcDataSourceDialect dataSourceDialect,
-            SqlServerConnection dataConnection,
-            SqlServerConnection metadataConnection) {
+            JdbcSourceConfig sourceConfig, JdbcDataSourceDialect dataSourceDialect) {
         super(sourceConfig, dataSourceDialect);
-        this.dataConnection = dataConnection;
-        this.metadataConnection = metadataConnection;
+
+        this.dataConnection = createSqlServerConnection(sourceConfig.getDbzConfiguration());
+        this.metadataConnection = createSqlServerConnection(sourceConfig.getDbzConfiguration());
         this.metadataProvider = new SqlServerEventMetadataProvider();
     }
 
@@ -158,6 +162,16 @@ public void configure(SourceSplitBase sourceSplitBase) {
         this.errorHandler = new SqlServerErrorHandler(connectorConfig.getLogicalName(), queue);
     }
 
+    @Override
+    public void close() {
+        try {
+            this.dataConnection.close();
+            this.metadataConnection.close();
+        } catch (SQLException e) {
+            log.warn("Failed to close connection, {}", ExceptionUtils.getMessage(e));
+        }
+    }
+
     @Override
     public SqlServerSourceConfig getSourceConfig() {
         return (SqlServerSourceConfig) sourceConfig;