[997] just load archive log when ORA-00310 exception happen

chunjun-connectors/chunjun-connector-binlog/src/main/java/com/dtstack/chunjun/connector/binlog/converter/BinlogColumnConverter.java

@@ -43,6 +43,7 @@
 import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
 import com.alibaba.otter.canal.protocol.CanalEntry;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -286,9 +287,18 @@ private void parseColumnList(
     @Override
     protected IDeserializationConverter createInternalConverter(String type) {
         String substring = type;
-        int index = type.indexOf(ConstantValue.LEFT_PARENTHESIS_SYMBOL);
+        // 为了支持无符号类型  如 int unsigned
+        if (StringUtils.contains(substring, ConstantValue.DATA_TYPE_UNSIGNED)) {
+            substring = substring.replaceAll(ConstantValue.DATA_TYPE_UNSIGNED, "").trim();
+        }
+
+        if (StringUtils.contains(substring, ConstantValue.DATA_TYPE_UNSIGNED_LOWER)) {
+            substring = substring.replaceAll(ConstantValue.DATA_TYPE_UNSIGNED_LOWER, "").trim();
+        }
+
+        int index = substring.indexOf(ConstantValue.LEFT_PARENTHESIS_SYMBOL);
         if (index > 0) {
-            substring = type.substring(0, index);
+            substring = substring.substring(0, index);
         }
         switch (substring.toUpperCase(Locale.ENGLISH)) {
             case "BIT":

chunjun-connectors/chunjun-connector-binlog/src/main/java/com/dtstack/chunjun/connector/binlog/listener/BinlogEventSink.java

@@ -83,8 +83,8 @@ public boolean sink(
                 processRowChange(rowChange, schema, table, executeTime);
             } catch (WriteRecordException e) {
                 // todo 脏数据记录
-                if (LOG.isTraceEnabled()) {
-                    LOG.trace(
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug(
                             "write error rowData, rowData = {}, e = {}",
                             e.getRowData().toString(),
                             ExceptionUtil.getErrorMessage(e));

chunjun-connectors/chunjun-connector-hive/src/main/java/com/dtstack/chunjun/connector/hive/util/HiveDbUtil.java

@@ -127,9 +127,7 @@ private static Connection getConnectionWithKerberos(
 
         UserGroupInformation ugi;
         try {
-            ugi =
-                    KerberosUtil.loginAndReturnUgi(
-                            conf.get(KerberosUtil.KEY_PRINCIPAL_FILE), principal, keytabFileName);
+            ugi = KerberosUtil.loginAndReturnUgi(conf, principal, keytabFileName);
         } catch (Exception e) {
             throw new RuntimeException("Login kerberos error:", e);
         }

chunjun-connectors/chunjun-connector-inceptor/pom.xml

@@ -233,6 +233,7 @@
 									<excludes>
 										<exclude>org.apache.hadoop.security.GroupMappingServiceProvider</exclude>
 										<exclude>org.apache.hadoop.fs.*</exclude>
+										<exclude>org.apache.hadoop.fs.permission.*</exclude>
 
 										<exclude>org.apache.hadoop.conf.*</exclude>
 
@@ -246,6 +247,8 @@
 										<exclude>org.apache.hadoop.filecache.*</exclude>
 
 										<exclude>org.apache.hadoop.io.retry.*</exclude>
+										<exclude>org.apache.hadoop.ipc.RemoteException</exclude>
+										<exclude>org.apache.hadoop.ipc.StandbyException</exclude>
 										<exclude>org.apache.hadoop.hdfs.server.*</exclude>
 										<exclude>org.apache.hadoop.hdfs.server.namenode.*</exclude>
 										<exclude>org.apache.hadoop.hdfs.server.namenode.ha.*</exclude>

chunjun-connectors/chunjun-connector-inceptor/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/ConfiguredFailoverProxyProvider.java

@@ -0,0 +1,134 @@
+//
+// Source code recreated from a .class file by IntelliJ IDEA
+// (powered by FernFlower decompiler)
+//
+
+package org.apache.hadoop.hdfs.server.namenode.ha;
+
+import com.google.common.base.Preconditions;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.HAUtil;
+import org.apache.hadoop.hdfs.NameNodeProxies;
+import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+public class ConfiguredFailoverProxyProvider<T> extends AbstractNNFailoverProxyProvider<T> {
+    private static final Log LOG = LogFactory.getLog(ConfiguredFailoverProxyProvider.class);
+    private final Configuration conf;
+    private final List<ConfiguredFailoverProxyProvider.AddressRpcProxyPair<T>> proxies =
+            new ArrayList();
+    private final UserGroupInformation ugi;
+    private final Class<T> xface;
+    private int currentProxyIndex = 0;
+
+    public ConfiguredFailoverProxyProvider(Configuration conf, URI uri, Class<T> xface) {
+        Preconditions.checkArgument(
+                xface.isAssignableFrom(NamenodeProtocols.class),
+                "Interface class %s is not a valid NameNode protocol!");
+        this.xface = xface;
+        this.conf = new Configuration(conf);
+        int maxRetries = this.conf.getInt("dfs.client.failover.connection.retries", 0);
+        this.conf.setInt("ipc.client.connect.max.retries", maxRetries);
+        int maxRetriesOnSocketTimeouts =
+                this.conf.getInt("dfs.client.failover.connection.retries.on.timeouts", 0);
+        this.conf.setInt("ipc.client.connect.max.retries.on.timeouts", maxRetriesOnSocketTimeouts);
+
+        try {
+            this.ugi = UserGroupInformation.getCurrentUser();
+            Map<String, Map<String, InetSocketAddress>> map = DFSUtil.getHaNnRpcAddresses(conf);
+            Map<String, InetSocketAddress> addressesInNN = (Map) map.get(uri.getHost());
+            if (addressesInNN != null && addressesInNN.size() != 0) {
+                Collection<InetSocketAddress> addressesOfNns = addressesInNN.values();
+                Iterator var9 = addressesOfNns.iterator();
+
+                while (var9.hasNext()) {
+                    InetSocketAddress address = (InetSocketAddress) var9.next();
+                    this.proxies.add(
+                            new ConfiguredFailoverProxyProvider.AddressRpcProxyPair(address));
+                }
+
+                HAUtil.cloneDelegationTokenForLogicalUri(this.ugi, uri, addressesOfNns);
+            } else {
+                throw new RuntimeException(
+                        "Could not find any configured addresses for URI " + uri);
+            }
+        } catch (IOException var11) {
+            throw new RuntimeException(var11);
+        }
+    }
+
+    public Class<T> getInterface() {
+        return this.xface;
+    }
+
+    public synchronized ProxyInfo<T> getProxy() {
+        ConfiguredFailoverProxyProvider.AddressRpcProxyPair<T> current =
+                (ConfiguredFailoverProxyProvider.AddressRpcProxyPair)
+                        this.proxies.get(this.currentProxyIndex);
+        if (current.namenode == null) {
+            try {
+                current.namenode =
+                        NameNodeProxies.createNonHAProxy(
+                                        this.conf,
+                                        current.address,
+                                        this.xface,
+                                        this.ugi,
+                                        false,
+                                        this.fallbackToSimpleAuth)
+                                .getProxy();
+            } catch (IOException var3) {
+                LOG.error("Failed to create RPC proxy to NameNode", var3);
+                throw new RuntimeException(var3);
+            }
+        }
+
+        return new ProxyInfo(current.namenode, current.address.toString());
+    }
+
+    public synchronized void performFailover(T currentProxy) {
+        this.currentProxyIndex = (this.currentProxyIndex + 1) % this.proxies.size();
+    }
+
+    public synchronized void close() throws IOException {
+        Iterator var1 = this.proxies.iterator();
+
+        while (var1.hasNext()) {
+            ConfiguredFailoverProxyProvider.AddressRpcProxyPair<T> proxy =
+                    (ConfiguredFailoverProxyProvider.AddressRpcProxyPair) var1.next();
+            if (proxy.namenode != null) {
+                if (proxy.namenode instanceof Closeable) {
+                    ((Closeable) proxy.namenode).close();
+                } else {
+                    RPC.stopProxy(proxy.namenode);
+                }
+            }
+        }
+    }
+
+    public boolean useLogicalURI() {
+        return true;
+    }
+
+    private static class AddressRpcProxyPair<T> {
+        public final InetSocketAddress address;
+        public T namenode;
+
+        public AddressRpcProxyPair(InetSocketAddress address) {
+            this.address = address;
+        }
+    }
+}

chunjun-connectors/chunjun-connector-mongodb/src/main/java/com/dtstack/chunjun/connector/mongodb/sink/MongodbOutputFormatBuilder.java

@@ -36,9 +36,11 @@
  */
 public class MongodbOutputFormatBuilder extends BaseRichOutputFormatBuilder {
     MongodbDataSyncConf mongodbDataSyncConf;
+    String upsertKey;
 
     public MongodbOutputFormatBuilder(MongodbDataSyncConf mongodbDataSyncConf) {
         this.mongodbDataSyncConf = mongodbDataSyncConf;
+        this.upsertKey = mongodbDataSyncConf.getReplaceKey();
         MongoClientConf mongoClientConf =
                 MongoClientConfFactory.createMongoClientConf(mongodbDataSyncConf);
         MongodbOutputFormat.WriteMode writeMode =
@@ -50,12 +52,12 @@ public MongodbOutputFormatBuilder(MongodbDataSyncConf mongodbDataSyncConf) {
 
     public MongodbOutputFormatBuilder(
             MongoClientConf mongoClientConf, String key, MongodbOutputFormat.WriteMode writeMode) {
+        this.upsertKey = key;
         this.format = new MongodbOutputFormat(mongoClientConf, key, writeMode);
     }
 
     @Override
     protected void checkFormat() {
-        String upsertKey = mongodbDataSyncConf.getReplaceKey();
         if (!StringUtils.isBlank(upsertKey)) {
             List<FieldConf> fields = mongodbDataSyncConf.getColumn();
             boolean flag = false;

chunjun-connectors/chunjun-connector-mongodb/src/main/java/com/dtstack/chunjun/connector/mongodb/table/lookup/MongoAllTableFunction.java

@@ -44,7 +44,7 @@
  */
 public class MongoAllTableFunction extends AbstractAllTableFunction {
 
-    private static final int FETCH_SIZE = 1000;
+    private final int fetchSize;
     private final MongoClientConf mongoClientConf;
     private transient MongoClient mongoClient;
     private transient MongoCollection collection;
@@ -57,6 +57,7 @@ public MongoAllTableFunction(
             String[] fieldNames) {
         super(fieldNames, keyNames, lookupConf, new MongodbRowConverter(rowType, fieldNames));
         this.mongoClientConf = mongoClientConf;
+        this.fetchSize = lookupConf.getFetchSize();
     }
 
     @Override
@@ -70,7 +71,7 @@ protected void loadData(Object cacheRef) {
         Map<String, List<Map<String, Object>>> tmpCache =
                 (Map<String, List<Map<String, Object>>>) cacheRef;
 
-        FindIterable<Document> findIterable = collection.find().limit(FETCH_SIZE);
+        FindIterable<Document> findIterable = collection.find().limit(fetchSize);
         MongoCursor<Document> mongoCursor = findIterable.iterator();
         while (mongoCursor.hasNext()) {
             Document doc = mongoCursor.next();

chunjun-connectors/chunjun-connector-mongodb/src/main/java/com/dtstack/chunjun/connector/mongodb/table/lookup/MongoLruTableFunction.java

@@ -41,8 +41,9 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * @author Ada Wong
@@ -86,7 +87,7 @@ public void handleAsyncInvoke(CompletableFuture<Collection<RowData>> future, Obj
             basicDbObject.append(keyNames[i], keys[i]);
         }
 
-        AtomicInteger atomicInteger = new AtomicInteger(0);
+        List<RowData> rowList = new CopyOnWriteArrayList<>();
 
         Block<Document> block =
                 (document) -> {
@@ -96,15 +97,16 @@ public void handleAsyncInvoke(CompletableFuture<Collection<RowData>> future, Obj
                     } catch (Exception e) {
                         LOG.error("", e);
                     }
-                    atomicInteger.incrementAndGet();
-                    future.complete(Collections.singleton(row));
+                    rowList.add(row);
                 };
 
         SingleResultCallback<Void> callbackWhenFinished =
                 (result, t) -> {
-                    if (atomicInteger.get() <= 0) {
+                    if (rowList.size() <= 0) {
                         LOG.warn("Cannot retrieve the data from the database");
                         future.complete(Collections.EMPTY_LIST);
+                    } else {
+                        future.complete(rowList);
                     }
                 };
 

chunjun-connectors/chunjun-connector-mongodb/src/main/java/com/dtstack/chunjun/connector/mongodb/table/lookup/MongodbAsyncTableFunction.java

@@ -40,8 +40,9 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * @author Ada Wong
@@ -87,20 +88,21 @@ public void eval(CompletableFuture<Collection<RowData>> future, Object... keys)
             basicDbObject.append(keyNames[i], keys[i]);
         }
 
-        AtomicInteger atomicInteger = new AtomicInteger(0);
+        List<RowData> rowList = new CopyOnWriteArrayList<>();
 
         Block<Document> block =
                 (document) -> {
                     RowData row = converter.toInternalLookup(document);
-                    atomicInteger.incrementAndGet();
-                    future.complete(Collections.singleton(row));
+                    rowList.add(row);
                 };
 
         SingleResultCallback<Void> callbackWhenFinished =
                 (result, t) -> {
-                    if (atomicInteger.get() <= 0) {
+                    if (rowList.size() <= 0) {
                         LOG.warn("Cannot retrieve the data from the database");
                         future.complete(Collections.EMPTY_LIST);
+                    } else {
+                        future.complete(rowList);
                     }
                 };
 

chunjun-connectors/chunjun-connector-oraclelogminer/src/main/java/com/dtstack/chunjun/connector/oraclelogminer/listener/LogFile.java

@@ -102,6 +102,10 @@ public void setType(String type) {
         this.type = type;
     }
 
+    public boolean isOnline() {
+        return "ONLINE".equals(this.type);
+    }
+
     @Override
     public String toString() {
         return "LogFile{"