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,8 +19,12 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import javax.sql.DataSource;
 
@@ -33,6 +37,7 @@
 import io.seata.core.model.BranchType;
 import io.seata.core.model.Resource;
 import io.seata.rm.DefaultResourceManager;
+import io.seata.rm.datasource.sql.struct.TableMetaCache;
 import io.seata.rm.datasource.sql.struct.TableMetaCacheFactory;
 import io.seata.rm.datasource.util.JdbcUtils;
 import io.seata.sqlparser.util.JdbcConstants;
@@ -53,6 +58,12 @@ public class DataSourceProxy extends AbstractDataSourceProxy implements Resource
 
     private static final String DEFAULT_RESOURCE_GROUP_ID = "DEFAULT";
 
+    private static final long TABLE_META_REFRESH_INTERVAL_TIME = 1000L;
+
+    private BlockingQueue<Long> tableMetaRefreshQueue;
+
+    private long lastRefreshTime;
+
     private String resourceGroupId;
 
     private String jdbcUrl;
@@ -77,9 +88,14 @@ public class DataSourceProxy extends AbstractDataSourceProxy implements Resource
     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,
+    private final ScheduledExecutorService tableMetaCheckExecutor = new ScheduledThreadPoolExecutor(1,
         new NamedThreadFactory("tableMetaChecker", 1, true));
 
+    private final Executor tableMetaRefreshExecutor = new ThreadPoolExecutor(1, 1,
+            0L, TimeUnit.MILLISECONDS,
+            new LinkedBlockingQueue<Runnable>(),
+            new NamedThreadFactory("tableMetaRefresh", 1, true));
+
     /**
      * Instantiates a new Data source proxy.
      *
@@ -106,6 +122,8 @@ public DataSourceProxy(DataSource targetDataSource, String resourceGroupId) {
 
     private void init(DataSource dataSource, String resourceGroupId) {
         this.resourceGroupId = resourceGroupId;
+        this.tableMetaRefreshQueue = new LinkedBlockingQueue<>();
+        this.lastRefreshTime = System.currentTimeMillis() - TABLE_META_REFRESH_INTERVAL_TIME;
         try (Connection connection = dataSource.getConnection()) {
             jdbcUrl = connection.getMetaData().getURL();
             dbType = JdbcUtils.getDbType(jdbcUrl);
@@ -121,19 +139,40 @@ 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);
+            tableMetaCheckExecutor.scheduleAtFixedRate(this::tableMetaRefreshEvent,
+                    0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
         }
-
+        tableMetaRefreshExecutor.execute(() -> {
+            while (true) {
+                try {
+                    Long eventTime = tableMetaRefreshQueue.take();
+                    // if it has bean refreshed not long ago, skip
+                    if (eventTime - lastRefreshTime > TABLE_META_REFRESH_INTERVAL_TIME) {
+                        try (Connection connection = dataSource.getConnection()) {
+                            TableMetaCache tableMetaCache = TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType());
+                            tableMetaCache.refresh(connection, DataSourceProxy.this.getResourceId());
+                        }
+                        lastRefreshTime = System.currentTimeMillis();
+                    }
+                } catch (Exception exx) {
+                    LOGGER.error("table refresh error:{}", exx.getMessage(), exx);
+                }
+            }
+        });
         //Set the default branch type to 'AT' in the RootContext.
         RootContext.setDefaultBranchType(this.getBranchType());
     }
 
+    /**
+     * public tableMeta refresh event
+     */
+    public void tableMetaRefreshEvent() {
+        boolean offer = tableMetaRefreshQueue.offer(System.currentTimeMillis());
+        if (!offer) {
+            LOGGER.error("table refresh event offer error:{}", resourceId);
+        }
+    }
+
     /**
      * 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/TableRecords.java

@@ -33,7 +33,11 @@
 import javax.sql.rowset.serial.SerialDatalink;
 import javax.sql.rowset.serial.SerialJavaObject;
 import javax.sql.rowset.serial.SerialRef;
+
+import com.google.common.collect.Interner;
+import com.google.common.collect.Interners;
 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;
@@ -55,6 +59,8 @@ public class TableRecords implements java.io.Serializable {
 
     private List<Row> rows = new ArrayList<Row>();
 
+    private static final Interner<String> TABLE_NAME_POOL = Interners.newWeakInterner();
+
     /**
      * Gets table name.
      *
@@ -196,7 +202,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 +265,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/sql/struct/TableRecordsTest.java

@@ -20,6 +20,7 @@
 import java.sql.Types;
 import java.util.List;
 
+import io.seata.rm.datasource.exception.TableMetaException;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -48,19 +49,36 @@ public class TableRecordsTest {
             new Object[] {"", "", "table_records_test", "description", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
         };
 
+    private static Object[][] columnMetasNewField =
+            new Object[][] {
+                    new Object[] {"", "", "table_records_test", "id", Types.INTEGER, "INTEGER", 64, 0, 10, 1, "", "", 0, 0, 64, 1, "NO", "YES"},
+                    new Object[] {"", "", "table_records_test", "name", Types.VARCHAR, "VARCHAR", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "information", Types.BLOB, "BLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "description", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+                    new Object[] {"", "", "table_records_test", "newf", Types.CLOB, "CLOB", 64, 0, 10, 0, "", "", 0, 0, 64, 2, "YES", "NO"},
+            };
+
     private static Object[][] indexMetas =
         new Object[][] {
             new Object[] {"PRIMARY", "id", false, "", 3, 1, "A", 34},
         };
 
     private static List<String> returnValueColumnLabels = Lists.newArrayList("id", "name", "information", "description");
 
+    private static List<String> returnValueColumnLabelsNewField = Lists.newArrayList("id", "name", "information", "description","newf");
+
     private static Object[][] returnValue =
         new Object[][] {
             new Object[] {1, "Tom", "hello", "world"},
             new Object[] {2, "Jack", "hello", "world"},
         };
 
+    private static Object[][] returnValueNewField =
+            new Object[][] {
+                    new Object[] {1, "Tom", "hello", "world","newf"},
+                    new Object[] {2, "Jack", "hello", "world","newf"},
+            };
+
     @BeforeEach
     public void initBeforeEach() {
     }
@@ -115,6 +133,25 @@ public void testBuildRecords() throws SQLException {
         Assertions.assertNotNull(tableRecords);
     }
 
+    @Test
+    public void testBuildRecordsNewFeild() throws SQLException {
+        MockDriver mockDriver = new MockDriver(returnValueColumnLabels, returnValue, columnMetas, indexMetas);
+        DruidDataSource dataSource = new DruidDataSource();
+        dataSource.setUrl("jdbc:mock:xxx");
+        dataSource.setDriver(mockDriver);
+        MockStatementBase mockStatement = new MockStatement(dataSource.getConnection().getConnection());
+        DataSourceProxy proxy = new DataSourceProxy(dataSource);
+
+        TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(JdbcConstants.MYSQL).getTableMeta(proxy.getPlainConnection(),
+                "table_records_test", proxy.getResourceId());
+
+
+        //  模拟新字段增加
+        MockDriver mockDriverNewField = new MockDriver(returnValueColumnLabelsNewField, returnValueNewField, columnMetasNewField, indexMetas);
+        ResultSet resultSet = mockDriverNewField.executeQuery(mockStatement, "select * from table_records_test");
+        Assertions.assertThrows(TableMetaException.class, () -> TableRecords.buildRecords(tableMeta, resultSet));
+    }
+
     @Test
     public void testEmpty() {
         TableRecords.EmptyTableRecords emptyTableRecords = new TableRecords.EmptyTableRecords();