Merge remote-tracking branch 'origin/v1.9.0_dev' into v1.10.0_dev

Author: todd5167

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

@@ -94,12 +94,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());
     }
 
@@ -123,9 +123,10 @@ public void flatMap(Tuple2<Boolean,Row> value, Collector<Tuple2<Boolean,Row>> ou
         List<Map<String, Object>> cacheList = cacheRef.get().get(cacheKey);
         if (CollectionUtils.isEmpty(cacheList) && sideInfo.getJoinType() == JoinType.LEFT) {
             out.collect(Tuple2.of(value.f0, fillData(value.f1, null)));
+        } else if (!CollectionUtils.isEmpty(cacheList)) {
+            cacheList.stream().forEach(one -> out.collect(Tuple2.of(value.f0, fillData(value.f1, one))));
         }
 
-        cacheList.stream().forEach(one -> out.collect(Tuple2.of(value.f0, fillData(value.f1, null))));
     }
 
     @Override
@@ -219,7 +220,7 @@ private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, C
             }
 
             String cacheKey = sideInfo.getEqualFieldList().stream()
-                    .map(equalField -> oneRow.get(equalField))
+                    .map(oneRow::get)
                     .map(Object::toString)
                     .collect(Collectors.joining("_"));
 
@@ -233,7 +234,8 @@ public int getFetchSize() {
     }
 
     /**
-     *  get jdbc connection
+     * get jdbc connection
+     *
      * @param dbURL
      * @param userName
      * @param password