Merge branch 'hotfix_1.8_3.10.x_27342' into 1.8_release_3.10.x

Author: gituser

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java

@@ -93,7 +93,6 @@ protected void initCache() throws SQLException {
     protected void reloadCache() {
         //reload cacheRef and replace to old cacheRef
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
-        cacheRef.set(newCache);
         try {
             loadData(newCache);
         } catch (SQLException e) {
@@ -123,6 +122,11 @@ public void flatMap(CRow value, Collector<CRow> out) throws Exception {
         List<Map<String, Object>> cacheList = cacheRef.get().get(cacheKey);
         if (CollectionUtils.isEmpty(cacheList) && sideInfo.getJoinType() == JoinType.LEFT) {
             out.collect(new CRow(fillData(value.row(), null), value.change()));
+            return;
+        }
+
+        if (CollectionUtils.isEmpty(cacheList)) {
+            return;
         }
 
         cacheList.forEach(one -> out.collect(new CRow(fillData(value.row(), one), value.change())));