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

Author: gituser

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

@@ -91,12 +91,12 @@ 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) {
             throw new RuntimeException(e);
         }
+        cacheRef.set(newCache);
         LOG.info("----- rdb all cacheRef reload end:{}", Calendar.getInstance());
     }
 
@@ -120,9 +120,9 @@ 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()));
+        } else if (!CollectionUtils.isEmpty(cacheList)) {
+            cacheList.forEach(one -> out.collect(new CRow(fillData(value.row(), one), value.change())));
         }
-
-        cacheList.stream().forEach(one -> out.collect(new CRow(fillData(value.row(), one), value.change())));
     }
 
     @Override
@@ -149,8 +149,8 @@ public Row fillData(Row input, Object sideInput) {
     }
 
     /**
-     *  covert flink time attribute.Type information for indicating event or processing time.
-     *  However, it behaves like a regular SQL timestamp but is serialized as Long.
+     * covert flink time attribute.Type information for indicating event or processing time.
+     * However, it behaves like a regular SQL timestamp but is serialized as Long.
      *
      * @param entry
      * @param obj
@@ -230,7 +230,8 @@ public int getFetchSize() {
     }
 
     /**
-     *  get jdbc connection
+     * get jdbc connection
+     *
      * @param dbURL
      * @param userName
      * @param password