feature: Introduce Cleanup API for TableMetaRefreshHolder Instance (#7559)

Author: YongGoose

changes/en-us/2.x.md

@@ -23,7 +23,8 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#7485](https://github.com/apache/incubator-seata/pull/7485)] Add http request filter for seata-server
 - [[#7509](https://github.com/apache/incubator-seata/pull/7509)] Reuse connection to merge branch transactions
 - [[#7492](https://github.com/apache/incubator-seata/pull/7492)] upgrade HTTP client in common module to support HTTP/2
-  [[#7551](https://github.com/apache/incubator-seata/pull/7551)] XAUtils add support for DM Database
+- [[#7551](https://github.com/apache/incubator-seata/pull/7551)] XAUtils add support for DM Database
+- [[#7559](https://github.com/apache/incubator-seata/pull/7559)] Introduce Cleanup API for TableMetaRefreshHolder Instance
 
 ### bugfix:
 

changes/zh-cn/2.x.md

@@ -24,6 +24,7 @@
 - [[#7509](https://github.com/apache/incubator-seata/pull/7509)] 复用连接合并分支事务
 - [[#7492](https://github.com/apache/incubator-seata/pull/7492)] 升级 common 模块中的 HTTP 客户端以支持 HTTP/2
 - [[#7551](https://github.com/apache/incubator-seata/pull/7551)] XAUtils支持达梦数据库
+- [[#7559](https://github.com/apache/incubator-seata/pull/7559)] 为 TableMetaRefreshHolder 实例引入清理 API
 
 
 ### bugfix:

rm-datasource/src/main/java/org/apache/seata/rm/datasource/DataSourceProxy.java

@@ -452,4 +452,9 @@ private void validMySQLVersion(Connection connection) {
             LOGGER.error("check mysql version fail error: {}", e.getMessage());
         }
     }
+
+    public void close() throws Exception {
+        // TODO: Need to unregister resource from DefaultResourceManager
+        TableMetaCacheFactory.shutdown(resourceId);
+    }
 }

rm-datasource/src/main/java/org/apache/seata/rm/datasource/sql/struct/TableMetaCacheFactory.java

@@ -108,7 +108,19 @@ private static void removeHolderFromMap(String resourceId) {
         LOGGER.info("Removed TableMetaRefreshHolder for resourceId: {}", resourceId);
     }
 
+    /**
+     * Shutdown all TableMetaRefreshHolder threads.
+     */
+    public static void shutdown(String resourceId) {
+        TableMetaRefreshHolder holder = TABLE_META_REFRESH_HOLDER_MAP.remove(resourceId);
+        if (holder != null) {
+            holder.shutdown();
+            LOGGER.info("TableMetaRefreshHolder for resourceId: {} has been shutdown.", resourceId);
+        }
+    }
+
     static class TableMetaRefreshHolder {
+        private volatile boolean stopped = false;
         private long lastRefreshFinishTime;
         private DataSourceProxy dataSource;
         private BlockingQueue<Long> tableMetaRefreshQueue;
@@ -128,7 +140,7 @@ static class TableMetaRefreshHolder {
             this.tableMetaRefreshQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE);
 
             tableMetaRefreshExecutor.execute(() -> {
-                while (true) {
+                while (!stopped) {
                     // 1. check table meta
                     if (ENABLE_TABLE_META_CHECKER_ENABLE
                             && System.nanoTime() - lastRefreshFinishTime
@@ -187,5 +199,13 @@ private boolean isDataSourceClosedException(SQLException ex) {
             }
             return StringUtils.isNotBlank(message) && message.contains("datasource") && message.contains("close");
         }
+
+        private void shutdown() {
+            stopped = true;
+            if (tableMetaRefreshExecutor instanceof ThreadPoolExecutor) {
+                ((ThreadPoolExecutor) tableMetaRefreshExecutor).shutdownNow();
+            }
+            LOGGER.info("TableMetaRefreshHolder shutdown for resourceId: {}", dataSource.getResourceId());
+        }
     }
 }

rm-datasource/src/test/java/org/apache/seata/rm/datasource/DataSourceProxyTest.java

@@ -17,6 +17,7 @@
 package org.apache.seata.rm.datasource;
 
 import com.alibaba.druid.pool.DruidDataSource;
+import org.apache.seata.rm.DefaultResourceManager;
 import org.apache.seata.rm.datasource.mock.MockDataSource;
 import org.apache.seata.rm.datasource.mock.MockDriver;
 import org.apache.seata.rm.datasource.sql.struct.TableMetaCacheFactory;
@@ -215,4 +216,28 @@ public static DataSourceProxy getDataSourceProxy(DataSource dataSource) {
             return new DataSourceProxy(dataSource);
         }
     }
+
+    @Test
+    public void testCloseRemovesResource() throws Exception {
+        final MockDriver mockDriver = new MockDriver();
+        final String username = "username";
+        final String jdbcUrl = "jdbc:mock:xxx";
+
+        final DruidDataSource dataSource = new DruidDataSource();
+        dataSource.setUrl(jdbcUrl);
+        dataSource.setDriver(mockDriver);
+        dataSource.setUsername(username);
+        dataSource.setPassword("password");
+
+        DataSourceProxy proxy = getDataSourceProxy(dataSource);
+        try (MockedStatic<DefaultResourceManager> drmStatic = Mockito.mockStatic(DefaultResourceManager.class);
+                MockedStatic<TableMetaCacheFactory> tmcfStatic = Mockito.mockStatic(TableMetaCacheFactory.class)) {
+            DefaultResourceManager drm = Mockito.mock(DefaultResourceManager.class);
+            drmStatic.when(DefaultResourceManager::get).thenReturn(drm);
+
+            proxy.close();
+
+            tmcfStatic.verify(() -> TableMetaCacheFactory.shutdown(proxy.getResourceId()));
+        }
+    }
 }

rm-datasource/src/test/java/org/apache/seata/rm/datasource/sql/struct/TableMetaCacheFactoryTest.java

@@ -17,19 +17,33 @@
 package org.apache.seata.rm.datasource.sql.struct;
 
 import org.apache.seata.common.loader.EnhancedServiceNotFoundException;
+import org.apache.seata.rm.datasource.DataSourceProxy;
+import org.apache.seata.rm.datasource.mock.MockDataSource;
 import org.apache.seata.rm.datasource.sql.struct.cache.MariadbTableMetaCache;
 import org.apache.seata.rm.datasource.sql.struct.cache.MysqlTableMetaCache;
 import org.apache.seata.rm.datasource.sql.struct.cache.OceanBaseTableMetaCache;
 import org.apache.seata.rm.datasource.sql.struct.cache.OracleTableMetaCache;
 import org.apache.seata.rm.datasource.sql.struct.cache.PolarDBXTableMetaCache;
 import org.apache.seata.sqlparser.util.JdbcConstants;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.lang.reflect.Field;
+import java.util.Map;
+
 public class TableMetaCacheFactoryTest {
 
     private static final String NOT_EXIST_SQL_TYPE = "not_exist_sql_type";
 
+    @BeforeEach
+    public void clearTableMetaRefreshHolderMap() throws Exception {
+        Field field = TableMetaCacheFactory.class.getDeclaredField("TABLE_META_REFRESH_HOLDER_MAP");
+        field.setAccessible(true);
+        Map<?, ?> map = (Map<?, ?>) field.get(null);
+        map.clear();
+    }
+
     @Test
     public void getTableMetaCache() {
         Assertions.assertTrue(
@@ -52,4 +66,23 @@ public void getTableMetaCache() {
             TableMetaCacheFactory.getTableMetaCache(NOT_EXIST_SQL_TYPE);
         });
     }
+
+    @Test
+    public void shutdownTest() throws NoSuchFieldException, IllegalAccessException {
+        DataSourceProxy dummy = new DataSourceProxy(new MockDataSource(), "dummy1");
+        TableMetaCacheFactory.registerTableMeta(dummy);
+
+        Map<String, TableMetaCacheFactory.TableMetaRefreshHolder> holderMap = getTableMetaRefreshHolderMap();
+        Assertions.assertEquals(1, holderMap.size());
+
+        TableMetaCacheFactory.shutdown("jdbc:mysql://127.0.0.1:3306/seata");
+        Assertions.assertTrue(holderMap.isEmpty(), "TableMetaRefreshHolder map should be empty after shutdown");
+    }
+
+    private Map<String, TableMetaCacheFactory.TableMetaRefreshHolder> getTableMetaRefreshHolderMap()
+            throws NoSuchFieldException, IllegalAccessException {
+        Field field = TableMetaCacheFactory.class.getDeclaredField("TABLE_META_REFRESH_HOLDER_MAP");
+        field.setAccessible(true);
+        return (Map<String, TableMetaCacheFactory.TableMetaRefreshHolder>) field.get(null);
+    }
 }