bugfix : check if table meta cache should be refreshed in AT mode

changes/en-us/develop.md

@@ -20,6 +20,7 @@ Add changes here for all PR submitted to the develop branch.
 - [[#5299](https://github.com/seata/seata/pull/5299)] fix GlobalSession deletion when retry rollback or retry commit timeout
 - [[#5307](https://github.com/seata/seata/pull/5307)] fix that keywords don't add escaped characters
 - [[#5311](https://github.com/seata/seata/pull/5311)] remove RollbackRetryTimeout sessions during in file storage recover
+- [[#4734](https://github.com/seata/seata/pull/4734)] check if table meta cache should be refreshed in AT mode
 - [[#5316](https://github.com/seata/seata/pull/5316)] fix G1 jvm parameter in jdk8
 - [[#5321](https://github.com/seata/seata/pull/5321)] fix When the rollback logic on the TC side returns RollbackFailed, the custom FailureHandler is not executed
 - [[#5332](https://github.com/seata/seata/pull/5332)] fix bugs found in unit tests

changes/zh-cn/develop.md

@@ -20,6 +20,7 @@
 - [[#5299](https://github.com/seata/seata/pull/5299)] 修复TC端重试回滚或重试提交超时GlobalSession的删除问题
 - [[#5307](https://github.com/seata/seata/pull/5307)] 修复生成update前后镜像sql不对关键字转义的bug
 - [[#5311](https://github.com/seata/seata/pull/5311)] 移除基于文件存储恢复时的RollbackRetryTimeout事务
+- [[#4734](https://github.com/seata/seata/pull/4734)] 修复AT模式下新增字段产生的字段找不到
 - [[#5316](https://github.com/seata/seata/pull/5316)] 修复jdk8 中 G1 参数
 - [[#5321](https://github.com/seata/seata/pull/5321)] 修复当TC端回滚返回RollbackFailed时，自定义FailureHandler的方法未执行
 - [[#5332](https://github.com/seata/seata/pull/5332)] 修复单元测试中发现的bug

rm-datasource/src/main/java/io/seata/rm/datasource/DataSourceProxy.java

@@ -19,15 +19,13 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+
 import javax.sql.DataSource;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import io.seata.common.Constants;
-import io.seata.common.thread.NamedThreadFactory;
-import io.seata.config.ConfigurationFactory;
-import io.seata.common.ConfigurationKeys;
 import io.seata.core.constants.DBType;
 import io.seata.core.context.RootContext;
 import io.seata.core.model.BranchType;
@@ -36,11 +34,6 @@
 import io.seata.rm.datasource.sql.struct.TableMetaCacheFactory;
 import io.seata.rm.datasource.util.JdbcUtils;
 import io.seata.sqlparser.util.JdbcConstants;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static io.seata.common.DefaultValues.DEFAULT_CLIENT_TABLE_META_CHECK_ENABLE;
-import static io.seata.common.DefaultValues.DEFAULT_TABLE_META_CHECKER_INTERVAL;
 
 /**
  * The type Data source proxy.
@@ -65,21 +58,6 @@ public class DataSourceProxy extends AbstractDataSourceProxy implements Resource
 
     private String version;
 
-    /**
-     * Enable the table meta checker
-     */
-    private static boolean ENABLE_TABLE_META_CHECKER_ENABLE = ConfigurationFactory.getInstance().getBoolean(
-        ConfigurationKeys.CLIENT_TABLE_META_CHECK_ENABLE, DEFAULT_CLIENT_TABLE_META_CHECK_ENABLE);
-
-    /**
-     * Table meta checker interval
-     */
-    private static final long TABLE_META_CHECKER_INTERVAL = ConfigurationFactory.getInstance().getLong(
-            ConfigurationKeys.CLIENT_TABLE_META_CHECKER_INTERVAL, DEFAULT_TABLE_META_CHECKER_INTERVAL);
-
-    private final ScheduledExecutorService tableMetaExecutor = new ScheduledThreadPoolExecutor(1,
-        new NamedThreadFactory("tableMetaChecker", 1, true));
-
     /**
      * Instantiates a new Data source proxy.
      *
@@ -120,20 +98,18 @@ private void init(DataSource dataSource, String resourceGroupId) {
         }
         initResourceId();
         DefaultResourceManager.get().registerResource(this);
-        if (ENABLE_TABLE_META_CHECKER_ENABLE) {
-            tableMetaExecutor.scheduleAtFixedRate(() -> {
-                try (Connection connection = dataSource.getConnection()) {
-                    TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())
-                        .refresh(connection, DataSourceProxy.this.getResourceId());
-                } catch (Exception ignore) {
-                }
-            }, 0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
-        }
-
+        TableMetaCacheFactory.registerTableMeta(this);
         //Set the default branch type to 'AT' in the RootContext.
         RootContext.setDefaultBranchType(this.getBranchType());
     }
 
+    /**
+     * publish tableMeta refresh event
+     */
+    public void tableMetaRefreshEvent() {
+        TableMetaCacheFactory.tableMetaRefreshEvent(this.getResourceId());
+    }
+
     /**
      * Gets plain connection.
      *

rm-datasource/src/main/java/io/seata/rm/datasource/exception/TableMetaException.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 1999-2019 Seata.io Group.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.seata.rm.datasource.exception;
+
+import java.sql.SQLException;
+
+/**
+ * The type TableMetaException exception.
+ *
+ * @author Bughue
+ */
+public class TableMetaException extends SQLException {
+    private String columnName;
+    private String tableName;
+
+    public TableMetaException(String tableName, String columnName) {
+        this.columnName = columnName;
+        this.tableName = tableName;
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    public String getColumnName() {
+        return columnName;
+    }
+}

rm-datasource/src/main/java/io/seata/rm/datasource/exec/AbstractDMLBaseExecutor.java

@@ -30,6 +30,7 @@
 import io.seata.rm.datasource.ConnectionContext;
 import io.seata.rm.datasource.ConnectionProxy;
 import io.seata.rm.datasource.StatementProxy;
+import io.seata.rm.datasource.exception.TableMetaException;
 import io.seata.rm.datasource.sql.struct.TableRecords;
 import io.seata.sqlparser.SQLRecognizer;
 import org.slf4j.Logger;
@@ -93,11 +94,18 @@ public T doExecute(Object... args) throws Throwable {
      * @throws Exception the exception
      */
     protected T executeAutoCommitFalse(Object[] args) throws Exception {
-        TableRecords beforeImage = beforeImage();
-        T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
-        TableRecords afterImage = afterImage(beforeImage);
-        prepareUndoLog(beforeImage, afterImage);
-        return result;
+        try {
+            TableRecords beforeImage = beforeImage();
+            T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
+            TableRecords afterImage = afterImage(beforeImage);
+            prepareUndoLog(beforeImage, afterImage);
+            return result;
+        } catch (TableMetaException e) {
+            LOGGER.error("table meta will be refreshed later, due to TableMetaException, table:{}, column:{}",
+                e.getTableName(), e.getColumnName());
+            statementProxy.getConnectionProxy().getDataSourceProxy().tableMetaRefreshEvent();
+            throw e;
+        }
     }
 
     private boolean isMultiPk() {

rm-datasource/src/main/java/io/seata/rm/datasource/exec/mysql/MySQLUpdateJoinExecutor.java

@@ -257,6 +257,7 @@ protected TableMeta getTableMeta(String tableName) {
      * @param afterImage  the after image
      * @return sql undo log
      */
+    @Override
     protected SQLUndoLog buildUndoItem(TableRecords beforeImage, TableRecords afterImage) {
         SQLType sqlType = sqlRecognizer.getSQLType();
         String tableName = beforeImage.getTableName();

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

@@ -15,19 +15,57 @@
  */
 package io.seata.rm.datasource.sql.struct;
 
+import java.sql.Connection;
 import java.util.Map;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
+import io.seata.common.ConfigurationKeys;
 import io.seata.common.loader.EnhancedServiceLoader;
+import io.seata.common.thread.NamedThreadFactory;
 import io.seata.common.util.CollectionUtils;
+import io.seata.config.ConfigurationFactory;
+import io.seata.rm.datasource.DataSourceProxy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static io.seata.common.DefaultValues.DEFAULT_CLIENT_TABLE_META_CHECK_ENABLE;
+import static io.seata.common.DefaultValues.DEFAULT_TABLE_META_CHECKER_INTERVAL;
 
 /**
+ * Table meta cache factory
+ *
  * @author guoyao
  */
 public class TableMetaCacheFactory {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(TableMetaCacheFactory.class);
+
     private static final Map<String, TableMetaCache> TABLE_META_CACHE_MAP = new ConcurrentHashMap<>();
 
+    private static final Map<String, TableMetaRefreshHolder> TABLE_META_REFRESH_HOLDER_MAP = new ConcurrentHashMap<>();
+
+    private static final long TABLE_META_REFRESH_INTERVAL_TIME = 1000L;
+
+    private static final int MAX_QUEUE_SIZE = 2000;
+
+    /**
+     * Enable the table meta checker
+     */
+    private static boolean ENABLE_TABLE_META_CHECKER_ENABLE = ConfigurationFactory.getInstance()
+        .getBoolean(ConfigurationKeys.CLIENT_TABLE_META_CHECK_ENABLE, DEFAULT_CLIENT_TABLE_META_CHECK_ENABLE);
+
+    /**
+     * Table meta checker interval
+     */
+    private static final long TABLE_META_CHECKER_INTERVAL = ConfigurationFactory.getInstance()
+        .getLong(ConfigurationKeys.CLIENT_TABLE_META_CHECKER_INTERVAL, DEFAULT_TABLE_META_CHECKER_INTERVAL);
+
+
     /**
      * get table meta cache
      *
@@ -38,4 +76,70 @@ public static TableMetaCache getTableMetaCache(String dbType) {
         return CollectionUtils.computeIfAbsent(TABLE_META_CACHE_MAP, dbType,
             key -> EnhancedServiceLoader.load(TableMetaCache.class, dbType));
     }
+
+    /**
+     * register table meta
+     *
+     * @param dataSourceProxy
+     */
+    public static void registerTableMeta(DataSourceProxy dataSourceProxy) {
+        TableMetaRefreshHolder holder = new TableMetaRefreshHolder(dataSourceProxy);
+        TABLE_META_REFRESH_HOLDER_MAP.put(dataSourceProxy.getResourceId(), holder);
+    }
+
+    /**
+     * public tableMeta refresh event
+     */
+    public static void tableMetaRefreshEvent(String resourceId) {
+        TableMetaRefreshHolder refreshHolder = TABLE_META_REFRESH_HOLDER_MAP.get(resourceId);
+        boolean offer = refreshHolder.tableMetaRefreshQueue.offer(System.currentTimeMillis());
+        if (!offer) {
+            LOGGER.error("table refresh event offer error:{}", resourceId);
+        }
+    }
+
+    static class TableMetaRefreshHolder {
+        private long lastRefreshFinishTime;
+        private DataSourceProxy dataSource;
+        private BlockingQueue<Long> tableMetaRefreshQueue;
+
+
+        private final Executor tableMetaRefreshExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
+            new LinkedBlockingQueue<>(), new NamedThreadFactory("tableMetaRefresh", 1, true));
+
+        TableMetaRefreshHolder(DataSourceProxy dataSource) {
+            this.dataSource = dataSource;
+            this.lastRefreshFinishTime = System.currentTimeMillis() - TABLE_META_REFRESH_INTERVAL_TIME;
+            this.tableMetaRefreshQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE);
+
+            tableMetaRefreshExecutor.execute(() -> {
+                while (true) {
+                    // 1. check table meta
+                    if (ENABLE_TABLE_META_CHECKER_ENABLE
+                        && System.currentTimeMillis() - lastRefreshFinishTime > TABLE_META_CHECKER_INTERVAL) {
+                        tableMetaRefreshEvent(dataSource.getResourceId());
+                    }
+
+                    // 2. refresh table meta
+                    try {
+                        Long eventTime = tableMetaRefreshQueue.take();
+                        // if it has bean refreshed not long ago, skip
+                        if (eventTime - lastRefreshFinishTime > TABLE_META_REFRESH_INTERVAL_TIME) {
+                            try (Connection connection = dataSource.getConnection()) {
+                                TableMetaCache tableMetaCache =
+                                    TableMetaCacheFactory.getTableMetaCache(dataSource.getDbType());
+                                tableMetaCache.refresh(connection, dataSource.getResourceId());
+                            }
+                            lastRefreshFinishTime = System.currentTimeMillis();
+                        }
+                    } catch (Exception exx) {
+                        LOGGER.error("table refresh error:{}", exx.getMessage(), exx);
+                    }
+                }
+            });
+        }
+
+
+
+    }
 }

rm-datasource/src/main/java/io/seata/rm/datasource/sql/struct/TableRecords.java

@@ -34,6 +34,7 @@
 import javax.sql.rowset.serial.SerialJavaObject;
 import javax.sql.rowset.serial.SerialRef;
 import io.seata.common.exception.ShouldNeverHappenException;
+import io.seata.rm.datasource.exception.TableMetaException;
 import io.seata.rm.datasource.sql.serial.SerialArray;
 import static io.seata.rm.datasource.exec.oracle.OracleJdbcType.TIMESTAMP_WITH_LOCAL_TIME_ZONE;
 import static io.seata.rm.datasource.exec.oracle.OracleJdbcType.TIMESTAMP_WITH_TIME_ZONE;
@@ -196,7 +197,7 @@ public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) th
             List<Field> fields = new ArrayList<>(columnCount);
             for (int i = 1; i <= columnCount; i++) {
                 String colName = resultSetMetaData.getColumnName(i);
-                ColumnMeta col = tmeta.getColumnMeta(colName);
+                ColumnMeta col = getColumnMeta(tmeta,colName);
                 int dataType = col.getDataType();
                 Field field = new Field();
                 field.setName(col.getColumnName());
@@ -259,6 +260,20 @@ public static TableRecords buildRecords(TableMeta tmeta, ResultSet resultSet) th
         return records;
     }
 
+    /**
+     * check if the column is null and return
+     *
+     * @param tmeta the table meta
+     * @param colName the column nmae
+     */
+    private static ColumnMeta getColumnMeta(TableMeta tmeta , String colName) throws SQLException {
+        ColumnMeta col = tmeta.getColumnMeta(colName);
+        if (col == null) {
+            throw new TableMetaException(tmeta.getTableName(), colName);
+        }
+        return col;
+    }
+
     /**
      * since there is no parameterless constructor for Blob, Clob and NClob just like mysql,
      * it needs to be converted to Serial_ type

rm-datasource/src/test/java/io/seata/rm/datasource/DataSourceProxyTest.java

@@ -22,6 +22,7 @@
 import com.alibaba.druid.pool.DruidDataSource;
 import io.seata.rm.datasource.mock.MockDataSource;
 import io.seata.rm.datasource.mock.MockDriver;
+import io.seata.rm.datasource.sql.struct.TableMetaCacheFactory;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -44,7 +45,7 @@ public void test_constructor() {
     @Test
     public void getResourceIdTest() throws SQLException, NoSuchFieldException, IllegalAccessException {
         // Disable 'DataSourceProxy.tableMetaExecutor' to prevent unit tests from being affected
-        Field enableField = DataSourceProxy.class.getDeclaredField("ENABLE_TABLE_META_CHECKER_ENABLE");
+        Field enableField = TableMetaCacheFactory.class.getDeclaredField("ENABLE_TABLE_META_CHECKER_ENABLE");
         enableField.setAccessible(true);
         enableField.set(null, false);