[hotfix-53323][logminer]Data is lost due to log group switchover during online log loading

Author: dujie

flinkx-connectors/flinkx-connector-oraclelogminer/src/main/java/com/dtstack/flinkx/connector/oraclelogminer/listener/LogMinerConnection.java

@@ -37,6 +37,7 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang.time.DateFormatUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -510,7 +511,7 @@ private void closeResources(ResultSet rs, Statement stmt, Connection conn) {
     }
 
     /** 根据leftScn 以及加载的日志大小限制 获取可加载的scn范围 以及此范围对应的日志文件 */
-    protected BigInteger getEndScn(BigInteger startScn, List<LogFile> logFiles)
+    protected Pair<BigInteger, Boolean> getEndScn(BigInteger startScn, List<LogFile> logFiles)
             throws SQLException {
 
         List<LogFile> logFileLists = new ArrayList<>();
@@ -557,6 +558,7 @@ protected BigInteger getEndScn(BigInteger startScn, List<LogFile> logFiles)
                                         .collect(Collectors.toList())));
 
         BigInteger endScn = startScn;
+        Boolean loadRedoLog = false;
 
         long fileSize = 0L;
         Collection<List<LogFile>> values = map.values();
@@ -598,10 +600,16 @@ protected BigInteger getEndScn(BigInteger startScn, List<LogFile> logFiles)
             // 解决logminer偶尔丢失数据问题，读取online日志的时候，需要将rightScn置为当前SCN
             endScn = getCurrentScn();
             logFiles = logFileLists;
+            // 如果加载了online日志  则loadRedoLog为true
+            loadRedoLog = true;
         }
 
-        LOG.info("getEndScn success,startScn:{},endScn:{}", startScn, endScn);
-        return endScn;
+        LOG.info(
+                "getEndScn success,startScn:{},endScn:{}, loadRedoLog:{}",
+                startScn,
+                endScn,
+                loadRedoLog);
+        return Pair.of(endScn, loadRedoLog);
     }
 
     /** 获取logminer加载的日志文件 */

flinkx-connectors/flinkx-connector-oraclelogminer/src/main/java/com/dtstack/flinkx/connector/oraclelogminer/listener/LogMinerHelper.java

@@ -25,6 +25,7 @@
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.commons.codec.DecoderException;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,6 +58,11 @@ public class LogMinerHelper {
     private final BigInteger step = new BigInteger("3000");
     private BigInteger startScn;
     private BigInteger endScn;
+    // 是否加载了online实时日志
+    private Boolean loadRedo = false;
+
+    // 最后一条数据的位点
+    private BigInteger currentSinkPosition;
     /** 当前正在读取的connection索引 * */
     private int currentIndex;
     /** 当前正在读取的connection * */
@@ -151,15 +157,26 @@ private void preLoad() throws SQLException {
                 // 按照加载日志文件大小限制，根据endScn作为起点找到对应的一组加载范围
                 BigInteger currentStartScn = Objects.nonNull(this.endScn) ? this.endScn : startScn;
 
-                BigInteger endScn =
+                // 如果加载了redo日志，则起点不能是上一次记载的日志的结束位点，而是上次消费的最后一条数据的位点
+                if (loadRedo) {
+                    // 需要加1  因为logminer查找数据是左闭右开，如果不加1  会导致最后一条数据重新消费
+                    currentStartScn = currentSinkPosition.add(BigInteger.ONE);
+                }
+
+                Pair<BigInteger, Boolean> endScn =
                         logMinerConnection.getEndScn(currentStartScn, new ArrayList<>(32));
-                logMinerConnection.startOrUpdateLogMiner(currentStartScn, endScn);
+                logMinerConnection.startOrUpdateLogMiner(currentStartScn, endScn.getLeft());
                 // 读取v$logmnr_contents 数据由线程池加载
                 loadData(logMinerConnection, logMinerSelectSql);
-                this.endScn = endScn;
+                this.endScn = endScn.getLeft();
+                this.loadRedo = endScn.getRight();
                 if (Objects.isNull(currentConnection)) {
                     updateCurrentConnection(logMinerConnection);
                 }
+                // 如果已经加载了redoLog就不需要多线程加载了
+                if (endScn.getRight()) {
+                    break;
+                }
             } else {
                 break;
             }
@@ -327,10 +344,15 @@ public LogMinerConnection chooseConnection() {
     }
 
     public QueueData getQueueData() {
-        return activeConnectionList.get(currentIndex).next();
+        QueueData next = activeConnectionList.get(currentIndex).next();
+        if (BigInteger.ZERO.compareTo(next.getScn()) != 0) {
+            this.currentSinkPosition = next.getScn();
+        }
+        return next;
     }
 
     public void setStartScn(BigInteger startScn) {
         this.startScn = startScn;
+        this.currentSinkPosition = this.startScn;
     }
 }