Merge branch 'v1.10.0_dev' into feat_1.10_impalaSinkKerberos_mergedDev

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java

Author: a49a

core/src/main/java/com/dtstack/flink/sql/constant/PluginParamConsts.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.constant;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/15
+ **/
+public class PluginParamConsts {
+    public static final String PRINCIPAL = "principal";
+    public static final String KEYTAB = "keytab";
+    public static final String KRB5_CONF = "krb5conf";
+}

core/src/main/java/com/dtstack/flink/sql/krb/KerberosTable.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.krb;
+
+import com.google.common.base.Strings;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/15
+ **/
+public interface KerberosTable {
+
+    String getPrincipal();
+
+    void setPrincipal(String principal);
+
+    String getKeytab();
+
+    void setKeytab(String keytab);
+
+    String getKrb5conf();
+
+    void setKrb5conf(String krb5conf);
+
+    boolean isEnableKrb();
+
+    void setEnableKrb(boolean enableKrb);
+
+    default void judgeKrbEnable() {
+        boolean allSet =
+            !Strings.isNullOrEmpty(getPrincipal()) &&
+            !Strings.isNullOrEmpty(getKeytab()) &&
+            !Strings.isNullOrEmpty(getKrb5conf());
+
+        boolean allNotSet =
+            Strings.isNullOrEmpty(getPrincipal()) &&
+            Strings.isNullOrEmpty(getKeytab()) &&
+            Strings.isNullOrEmpty(getKrb5conf());
+
+        if (allSet) {
+            setEnableKrb(true);
+        } else if (allNotSet) {
+            setEnableKrb(false);
+        } else {
+            throw new RuntimeException("Missing kerberos parameter! all kerberos params must be set, or all kerberos params are not set");
+        }
+    }
+}

core/src/main/java/com/dtstack/flink/sql/parser/SqlParser.java

@@ -28,8 +28,11 @@
 import com.google.common.collect.Lists;
 import com.google.common.base.Strings;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Reason:
@@ -51,6 +54,8 @@ public static void setLocalSqlPluginRoot(String localSqlPluginRoot){
         LOCAL_SQL_PLUGIN_ROOT = localSqlPluginRoot;
     }
 
+    private static final Pattern ADD_FIlE_PATTERN = Pattern.compile("(?i).*add\\s+file\\s+.+");
+
     /**
      * flink support sql syntax
      * CREATE TABLE sls_stream() with ();
@@ -70,6 +75,7 @@ public static SqlTree parseSql(String sql, String pluginLoadMode) throws Excepti
                 .replace("\t", " ").trim();
 
         List<String> sqlArr = DtStringUtil.splitIgnoreQuota(sql, SQL_DELIMITER);
+        sqlArr = removeAddFileStmt(sqlArr);
         SqlTree sqlTree = new SqlTree();
         AbstractTableInfoParser tableInfoParser = new AbstractTableInfoParser();
         for(String childSql : sqlArr){
@@ -150,4 +156,18 @@ public static SqlTree parseSql(String sql, String pluginLoadMode) throws Excepti
 
         return sqlTree;
     }
+
+    /**
+     * remove add file with statment etc. add file /etc/krb5.conf;
+     */
+    private static List<String> removeAddFileStmt(List<String> stmts) {
+        List<String> cleanedStmts = new ArrayList<>();
+        for (String stmt : stmts) {
+            Matcher matcher = ADD_FIlE_PATTERN.matcher(stmt);
+            if(!matcher.matches()) {
+                cleanedStmts.add(stmt);
+            }
+        }
+        return cleanedStmts;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/side/SidePredicatesParser.java

@@ -140,7 +140,7 @@ private void extractPredicateInfo(SqlNode whereNode, List<PredicateInfo> predica
         }
     }
 
-    private void fillPredicateInfoToList(SqlBasicCall whereNode, List<PredicateInfo> predicatesInfoList, String operatorName, SqlKind operatorKind,
+    private void  fillPredicateInfoToList(SqlBasicCall whereNode, List<PredicateInfo> predicatesInfoList, String operatorName, SqlKind operatorKind,
                                          int fieldIndex, int conditionIndex) {
         SqlNode sqlNode = whereNode.getOperands()[fieldIndex];
         if (sqlNode.getKind() == SqlKind.IDENTIFIER) {

core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java

@@ -50,4 +50,5 @@ public static UserGroupInformation getUgi(String principal, String keytabPath, S
         UserGroupInformation.setConfiguration(configuration);
         return UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytabPath);
     }
+
 }

core/src/test/java/com/dtstack/flink/sql/parser/SqlParserTest.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.parser;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.powermock.reflect.Whitebox;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @program: flink.sql
+ * @author: wuren
+ * @create: 2020/09/15
+ **/
+public class SqlParserTest {
+
+    @Test
+    public void testRemoveAddFileStmt() throws Exception {
+        List<String> rawStmts = new ArrayList<>();
+        String sql1 = "        add    file     asdasdasd   ";
+        String sql2 = "        aDd    fIle    With    asdasdasd   ";
+        String sql3 = "        INSERT INTO dwd_foo SELECT id, name FROM ods_foo";
+        String sql4 = "        ADD  FILE   asb    ";
+        rawStmts.add(sql1);
+        rawStmts.add(sql2);
+        rawStmts.add(sql3);
+        rawStmts.add(sql4);
+
+        List<String> stmts = Whitebox.invokeMethod(SqlParser.class, "removeAddFileStmt", rawStmts);
+        Assert.assertEquals(stmts.get(0), sql3);
+    }
+
+}

docs/plugin/kuduSide.md

@@ -60,6 +60,11 @@
 | isFaultTolerant |查询是否容错  查询失败是否扫描第二个副本  默认false  容错 | 否||
 | cache | 维表缓存策略(NONE/LRU/ALL)|否|NONE|
 | partitionedJoin | 是否在維表join之前先根据 設定的key 做一次keyby操作(可以減少维表的数据缓存量)|否|false|
+| principal |kerberos用于登录的principal | 否||
+| keytab |keytab文件的路径 | 否||
+| krb5conf |conf文件路径 | 否||
+Kerberos三个参数全部设置则开启Kerberos认证，如果缺少任何一个则会提示缺少参数错误。
+如果全部未设置则不开启Kerberos连接Kudu集群。
 --------------
 
 ## 5.样例
@@ -163,3 +168,20 @@ into
     on t1.id = t2.id;     
 ```
 
+## 7.kerberos示例
+```
+create table dim (
+    name varchar,
+    id int,
+    PERIOD FOR SYSTEM_TIME
+) WITH (
+    type='kudu',
+    kuduMasters='host1',
+    tableName='foo',
+    parallelism ='1',
+    cache ='ALL',
+    keytab='foo/foobar.keytab',
+    krb5conf='bar/krb5.conf',
+    principal='kudu/host1@DTSTACK.COM'
+);
+```

docs/plugin/kuduSink.md

@@ -42,7 +42,11 @@ kudu 1.9.0+cdh6.2.0
 | defaultOperationTimeoutMs | 操作超时时间 |否|
 | defaultSocketReadTimeoutMs | socket读取超时时间 |否|
 | parallelism | 并行度设置|否|1|
-      
+| principal |kerberos用于登录的principal | 否||
+| keytab |keytab文件的路径 | 否||
+| krb5conf |conf文件路径 | 否||
+Kerberos三个参数全部设置则开启Kerberos认证，如果缺少任何一个则会提示缺少参数错误。
+如果全部未设置则不开启Kerberos连接Kudu集群。      
 
 ## 5.样例：
 ```
@@ -123,4 +127,21 @@ into
 ### 结果数据
 ```
 {"a":"2","b":"2","c":"3","d":"4"}
-```
+```
+
+## 7.kerberos示例
+```
+create table dwd (
+    name varchar,
+    id int
+) WITH (
+    type='kudu',
+    kuduMasters='host1',
+    tableName='foo',
+    writeMode='insert',
+    parallelism ='1',
+    keytab='foo/foobar.keytab',
+    krb5conf='bar/krb5.conf',
+    principal='kudu/host1@DTSTACK.COM'
+);
+```

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java

@@ -54,6 +54,8 @@ protected Properties getKafkaProperties(KafkaSourceTableInfo kafkaSourceTableInf
 
         if (DtStringUtil.isJson(kafkaSourceTableInfo.getOffsetReset())) {
             props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.NONE.name().toLowerCase());
+        } else if(StringUtils.equalsIgnoreCase(EKafkaOffset.TIMESTAMP.name().toLowerCase(),  kafkaSourceTableInfo.getOffsetReset())){
+                props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.EARLIEST.name().toLowerCase());
         } else {
             props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSourceTableInfo.getOffsetReset());
         }
@@ -95,10 +97,12 @@ protected void setParallelism(Integer parallelism, DataStreamSource kafkaSource)
         }
     }
 
-    protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc) {
+    protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc, Runnable runnable) {
         if (StringUtils.equalsIgnoreCase(offset, EKafkaOffset.EARLIEST.name())) {
             kafkaSrc.setStartFromEarliest();
-        } else if (DtStringUtil.isJson(offset)) {
+        } else if(StringUtils.equalsIgnoreCase(offset, EKafkaOffset.TIMESTAMP.name())) {
+            runnable.run();
+        }else if (DtStringUtil.isJson(offset)) {
             Map<KafkaTopicPartition, Long> specificStartupOffsets = buildOffsetMap(offset, topicName);
             kafkaSrc.setStartFromSpecificOffsets(specificStartupOffsets);
         } else {

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/enums/EKafkaOffset.java

@@ -27,5 +27,6 @@ public enum EKafkaOffset {
 
     LATEST,
     EARLIEST,
+    TIMESTAMP,
     NONE
 }