[feat] add restapi

Author: yanghuaiGit

flinkx-binlog/pom.xml

@@ -23,6 +23,16 @@
             <artifactId>flinkx-core</artifactId>
             <version>1.6</version>
             <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-classic</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>ch.qos.logback</groupId>
+                    <artifactId>logback-core</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
     </dependencies>
 

flinkx-core/pom.xml

@@ -26,8 +26,8 @@
 
         <dependency>
             <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
-            <version>1.7.20</version>
+            <artifactId>slf4j-log4j12</artifactId>
+            <version>1.7.10</version>
         </dependency>
 
         <dependency>

flinkx-core/src/main/java/com/dtstack/flinkx/config/AbstractConfig.java

@@ -18,13 +18,15 @@
 
 package com.dtstack.flinkx.config;
 
+import com.dtstack.flinkx.util.GsonUtil;
 import com.google.gson.internal.LinkedTreeMap;
 
 import java.io.Serializable;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 
 /**
  * Abstract Config
@@ -129,7 +131,7 @@ public int getIntVal(String key, int defaultValue) {
         if(ret instanceof BigDecimal) {
             return ((BigDecimal)ret).intValue();
         }
-        throw new RuntimeException("can't cast " + key + " from " + ret.getClass().getName() + " to Integer");
+        throw new RuntimeException(String.format("cant't %s from %s to int, internalMap = %s", key, ret.getClass().getName(), GsonUtil.GSON.toJson(internalMap)));
     }
 
     public long getLongVal(String key, long defaultValue) {
@@ -158,7 +160,7 @@ public long getLongVal(String key, long defaultValue) {
         if(ret instanceof BigDecimal) {
             return ((BigDecimal)ret).longValue();
         }
-        throw new RuntimeException("can't cast " + key + " from " + ret.getClass().getName() + " to Long");
+        throw new RuntimeException(String.format("cant't %s from %s to long, internalMap = %s", key, ret.getClass().getName(), GsonUtil.GSON.toJson(internalMap)));
     }
 
     public double getDoubleVal(String key, double defaultValue) {
@@ -187,7 +189,7 @@ public double getDoubleVal(String key, double defaultValue) {
         if (ret instanceof BigDecimal) {
             return ((BigDecimal) ret).doubleValue();
         }
-        throw new RuntimeException("can't cast " + key + " from " + ret.getClass().getName() + " to Long");
+        throw new RuntimeException(String.format("cant't %s from %s to double, internalMap = %s", key, ret.getClass().getName(), GsonUtil.GSON.toJson(internalMap)));
     }
 
 
@@ -199,7 +201,33 @@ public boolean getBooleanVal(String key, boolean defaultValue) {
         if (ret instanceof Boolean) {
             return (Boolean) ret;
         }
-        throw new RuntimeException("can't cast " + key + " from " + ret.getClass().getName() + " to Long");
+        throw new RuntimeException(String.format("cant't %s from %s to boolean, internalMap = %s", key, ret.getClass().getName(), GsonUtil.GSON.toJson(internalMap)));
+    }
+
+    /**
+     * 从指定key中获取Properties配置信息
+     * @param key
+     * @param p
+     * @return
+     */
+    @SuppressWarnings("unchecked")
+    public Properties getProperties(String key, Properties p ){
+        Object ret = internalMap.get(key);
+        if(p == null){
+            p = new Properties();
+        }
+        if (ret == null) {
+            return p;
+        }
+        if(ret instanceof Map){
+            Map<String, Object> map = (Map<String, Object>) ret;
+            for (Map.Entry<String, Object> entry : map.entrySet()) {
+                p.setProperty(entry.getKey(), String.valueOf(entry.getValue()));
+            }
+            return p;
+        }else{
+            throw new RuntimeException(String.format("cant't %s from %s to map, internalMap = %s", key, ret.getClass().getName(), GsonUtil.GSON.toJson(internalMap)));
+        }
     }
 
 }

flinkx-core/src/main/java/com/dtstack/flinkx/constants/ConstantValue.java

@@ -27,6 +27,7 @@ public class ConstantValue {
 
     public static final String STAR_SYMBOL = "*";
     public static final String POINT_SYMBOL = ".";
+    public static final String TWO_POINT_SYMBOL = "..";
     public static final String EQUAL_SYMBOL = "=";
     public static final String SINGLE_QUOTE_MARK_SYMBOL = "'";
     public static final String DOUBLE_QUOTE_MARK_SYMBOL = "\"";
@@ -38,6 +39,10 @@ public class ConstantValue {
     public static final String LEFT_PARENTHESIS_SYMBOL = "(";
     public static final String RIGHT_PARENTHESIS_SYMBOL = ")";
 
+
+    public static final String DATA_TYPE_UNSIGNED = "UNSIGNED";
+
+
     public static final String KEY_HTTP = "http";
 
     public static final String PROTOCOL_HTTP = "http://";

flinkx-core/src/main/java/com/dtstack/flinkx/enums/ColumnType.java

@@ -19,9 +19,11 @@
 package com.dtstack.flinkx.enums;
 
 import com.dtstack.flinkx.constants.ConstantValue;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 
 /**
  * Define standard column type for all the readers or writers that do not
@@ -66,6 +68,12 @@ public enum ColumnType {
             STRING, VARCHAR, VARCHAR2, CHAR, NVARCHAR, TEXT, KEYWORD, BINARY
     );
 
+    /**
+     * 根据字段类型的字符串找出对应的枚举
+     * 找不到直接报错 IllegalArgumentException
+     * @param type
+     * @return
+     */
     public static ColumnType fromString(String type) {
         if(type == null) {
             throw new RuntimeException("null ColumnType!");
@@ -75,15 +83,31 @@ public static ColumnType fromString(String type) {
             type = type.substring(0, type.indexOf(ConstantValue.LEFT_PARENTHESIS_SYMBOL));
         }
 
-        return valueOf(type.toUpperCase());
+        type =  type.toUpperCase(Locale.ENGLISH);
+        //为了支持无符号类型  如 int unsigned
+        if(StringUtils.contains(type,ConstantValue.DATA_TYPE_UNSIGNED)){
+            type = type.replace(ConstantValue.DATA_TYPE_UNSIGNED,"").trim();
+        }
+        return valueOf(type);
     }
 
+    /**
+     * 根据字段类型的字符串找到对应的枚举 找不到就直接返回ColumnType.STRING;
+     * @param type
+     * @return
+     */
     public static ColumnType getType(String type){
+        type = type.toUpperCase(Locale.ENGLISH);
         if(type.contains(ConstantValue.LEFT_PARENTHESIS_SYMBOL)){
             type = type.substring(0, type.indexOf(ConstantValue.LEFT_PARENTHESIS_SYMBOL));
         }
 
-        if(type.toLowerCase().contains(ColumnType.TIMESTAMP.name().toLowerCase())){
+        //为了支持无符号类型  如 int unsigned
+        if(StringUtils.contains(type,ConstantValue.DATA_TYPE_UNSIGNED)){
+            type = type.replaceAll(ConstantValue.DATA_TYPE_UNSIGNED,"").trim();
+        }
+
+        if(type.contains(ColumnType.TIMESTAMP.name())){
             return TIMESTAMP;
         }
 

flinkx-core/src/main/java/com/dtstack/flinkx/latch/MetricLatch.java

@@ -19,6 +19,7 @@
 package com.dtstack.flinkx.latch;
 
 import com.dtstack.flinkx.constants.ConstantValue;
+import com.dtstack.flinkx.util.GsonUtil;
 import com.dtstack.flinkx.util.UrlUtil;
 import com.google.gson.Gson;
 import com.google.gson.internal.LinkedTreeMap;
@@ -76,10 +77,11 @@ private int getIntMetricVal(String requestUrl) {
         try(InputStream inputStream = UrlUtil.open(requestUrl)) {
             try(Reader rd = new InputStreamReader(inputStream, StandardCharsets.UTF_8)) {
                 Map<String,Object> map = gson.fromJson(rd, Map.class);
+                LOG.info("requestUrl = {}, and return map = {}", requestUrl, GsonUtil.GSON.toJson(map));
                 List<LinkedTreeMap> userTaskAccumulators = (List<LinkedTreeMap>) map.get("user-task-accumulators");
                 for(LinkedTreeMap accumulator : userTaskAccumulators) {
                     if(metricName != null && metricName.equals(accumulator.get("name"))) {
-                        return Integer.valueOf((String )accumulator.get("value"));
+                        return Integer.parseInt((String )accumulator.get("value"));
                     }
                 }
             } catch (Exception e) {

flinkx-core/src/main/java/com/dtstack/flinkx/metrics/BaseMetric.java

@@ -56,7 +56,7 @@ public void addMetric(String metricName, LongCounter counter, boolean meterView)
         metricCounters.put(metricName, counter);
         flinkxOutput.gauge(metricName, new SimpleAccumulatorGauge<>(counter));
         if (meterView){
-            flinkxOutput.meter(metricName + Metrics.SUFFIX_RATE, new SimpleLongCounterMeterView(counter, 60));
+            flinkxOutput.meter(metricName + Metrics.SUFFIX_RATE, new SimpleLongCounterMeterView(counter, 20));
         }
     }
 

flinkx-core/src/main/java/com/dtstack/flinkx/options/Options.java

@@ -82,6 +82,15 @@ public class Options {
     @OptionRequired(description = "plugin load mode, by classpath or shipfile")
     private String pluginLoadMode = "shipfile";
 
+    @OptionRequired(description = "kerberos krb5conf")
+    private String krb5conf ;
+
+    @OptionRequired(description = "kerberos keytabPath")
+    private String keytab ;
+
+    @OptionRequired(description = "kerberos principal")
+    private String principal ;
+
     @OptionRequired(description = "applicationId on yarn cluster")
     private String appId;
 
@@ -248,6 +257,30 @@ public void setP(String p) {
         this.p = p;
     }
 
+    public String getKrb5conf() {
+        return krb5conf;
+    }
+
+    public void setKrb5conf(String krb5conf) {
+        this.krb5conf = krb5conf;
+    }
+
+    public String getKeytab() {
+        return keytab;
+    }
+
+    public void setKeytab(String keytab) {
+        this.keytab = keytab;
+    }
+
+    public String getPrincipal() {
+        return principal;
+    }
+
+    public void setPrincipal(String principal) {
+        this.principal = principal;
+    }
+
     @Override
     public String toString() {
         return "Options{" +
@@ -268,7 +301,9 @@ public String toString() {
                 ", pluginLoadMode='" + pluginLoadMode + '\'' +
                 ", appId='" + appId + '\'' +
                 ", remotePluginPath='" + remotePluginPath + '\'' +
-                ", flinkConfiguration=" + flinkConfiguration +
+                ", krb5conf='" + krb5conf + '\'' +
+                ", keytab='" + keytab + '\'' +
+                ", principal='" + principal + '\'' +
                 '}';
     }
 }

flinkx-core/src/main/java/com/dtstack/flinkx/outputformat/BaseFileOutputFormat.java

@@ -106,7 +106,7 @@ protected void openInternal(int taskNumber, int numTasks) throws IOException {
         nextBlock();
     }
 
-    private void initPath(){
+    protected void initPath(){
         if(StringUtils.isNotBlank(fileName)) {
             outputFilePath = path + SP + fileName;
         } else {
@@ -122,7 +122,7 @@ private void initPath(){
                 taskNumber, currentBlockFileNamePrefix, tmpPath, finishedPath);
     }
 
-    private void initFileIndex() {
+    protected void initFileIndex() {
         if (null != formatState && formatState.getFileIndex() > -1) {
             blockIndex = formatState.getFileIndex() + 1;
         }

flinkx-core/src/main/java/com/dtstack/flinkx/util/FileSystemUtil.java

@@ -69,8 +69,9 @@ public static void setHadoopUserName(Configuration conf){
         }
 
         try {
-            String ticketCachePath = conf.get("hadoop.security.kerberos.ticket.cache.path");
-            UserGroupInformation ugi = UserGroupInformation.getBestUGI(ticketCachePath, hadoopUserName);
+            String previousUserName = UserGroupInformation.getLoginUser().getUserName();
+            LOG.info("Hadoop user from '{}' switch to '{}' with SIMPLE auth", previousUserName, hadoopUserName);
+            UserGroupInformation ugi = UserGroupInformation.createRemoteUser(hadoopUserName);
             UserGroupInformation.setLoginUser(ugi);
         } catch (Exception e) {
             LOG.warn("Set hadoop user name error:", e);
@@ -87,7 +88,6 @@ public static boolean isOpenKerberos(Map<String, Object> hadoopConfig){
 
     private static FileSystem getFsWithKerberos(Map<String, Object> hadoopConfig, String defaultFs) throws Exception{
         UserGroupInformation ugi = getUGI(hadoopConfig, defaultFs);
-        UserGroupInformation.setLoginUser(ugi);
 
         return ugi.doAs(new PrivilegedAction<FileSystem>() {
             @Override
@@ -110,7 +110,6 @@ public static UserGroupInformation getUGI(Map<String, Object> hadoopConfig, Stri
         KerberosUtil.refreshConfig();
 
         UserGroupInformation ugi = KerberosUtil.loginAndReturnUgi(getConfiguration(hadoopConfig, defaultFs), principal, keytabFileName);
-        UserGroupInformation.setLoginUser(ugi);
 
         return ugi;
     }