Merge branch 'feat_1.10_kuduKerberos' into '1.10_test_4.1.x'

Feat 1.10 kudu kerberos

See merge request dt-insight-engine/flinkStreamSQL!98

Author: zoudaokoulife

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/util/KrbUtils.java

@@ -0,0 +1,54 @@
+/*
+ * 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.util;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/14
+ **/
+public class KrbUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KrbUtils.class);
+
+    public static final String KRB5_CONF_KEY = "java.security.krb5.conf";
+    public static final String HADOOP_AUTH_KEY = "hadoop.security.authentication";
+    public static final String KRB_STR = "Kerberos";
+//    public static final String FALSE_STR = "false";
+//    public static final String SUBJECT_ONLY_KEY = "javax.security.auth.useSubjectCredsOnly";
+
+    public static UserGroupInformation getUgi(String principal, String keytabPath, String krb5confPath) throws IOException {
+        LOG.info("Kerberos login with principal: {} and keytab: {}", principal, keytabPath);
+        System.setProperty(KRB5_CONF_KEY, krb5confPath);
+        // TODO 尚未探索出此选项的意义，以后研究明白方可打开
+//        System.setProperty(SUBJECT_ONLY_KEY, FALSE_STR);
+        Configuration configuration = new Configuration();
+        configuration.set(HADOOP_AUTH_KEY , KRB_STR);
+        UserGroupInformation.setConfiguration(configuration);
+        return UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytabPath);
+    }
+
+}

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
 }

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

@@ -56,6 +56,9 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));
         kafkaSourceTableInfo.setSourceDataType(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.SOURCE_DATA_TYPE_KEY.toLowerCase(), FormatType.DT_NEST.name())));
 
+        if(props.containsKey(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase())){
+            kafkaSourceTableInfo.setTimestampOffset(MathUtil.getLongVal(props.getOrDefault(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase(), System.currentTimeMillis())));
+        }
         Map<String, String> kafkaParams = props.keySet().stream()
                 .filter(key -> !key.isEmpty() && key.startsWith("kafka."))
                 .collect(Collectors.toMap(

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

@@ -55,6 +55,8 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public static final String CHARSET_NAME_KEY = "charsetName";
 
+    public static final String TIMESTAMP_OFFSET = "timestampOffset";
+
     private String bootstrapServers;
 
     private String topic;
@@ -75,6 +77,8 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public String charsetName;
 
+    private Long timestampOffset;
+
     public String getBootstrapServers() {
         return bootstrapServers;
     }
@@ -159,6 +163,14 @@ public void setCharsetName(String charsetName) {
 		this.charsetName = charsetName;
 	}
 
+    public Long getTimestampOffset() {
+        return timestampOffset;
+    }
+
+    public void setTimestampOffset(Long timestampOffset) {
+        this.timestampOffset = timestampOffset;
+    }
+
 	@Override
 	public boolean check() {
 		Preconditions.checkNotNull(getType(), "kafka of type is required");

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

@@ -51,7 +51,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 
         setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
-        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
+        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromTimestamp(kafkaSourceTableInfo.getTimestampOffset()));
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);