complete kudu kerberos

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

@@ -0,0 +1,52 @@
+/*
+ * 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 java.io.IOException;
+
+/**
+ * @program: flinkStreamSQL
+ * @author: wuren
+ * @create: 2020/09/14
+ **/
+public class KrbUtils {
+
+    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 {
+        System.setProperty(KRB5_CONF_KEY, krb5confPath);
+//        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);
+    }
+
+    public void checkKrbParams() {
+
+    }
+
+}

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'
+);
+```

kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java

@@ -7,6 +7,7 @@
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.kudu.table.KuduSideTableInfo;
 import com.dtstack.flink.sql.side.kudu.utils.KuduUtil;
+import com.dtstack.flink.sql.util.KrbUtils;
 import com.dtstack.flink.sql.util.RowDataComplete;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
@@ -18,6 +19,7 @@
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.kudu.ColumnSchema;
 import org.apache.kudu.Schema;
 import org.apache.kudu.client.KuduClient;
@@ -31,6 +33,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.security.PrivilegedAction;
 import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.Calendar;
@@ -211,24 +215,8 @@ private String buildKey(Map<String, Object> val, List<String> equalFieldList) {
     private KuduScanner getConn(KuduSideTableInfo tableInfo) {
         try {
             if (client == null) {
-                String kuduMasters = tableInfo.getKuduMasters();
                 String tableName = tableInfo.getTableName();
-                Integer workerCount = tableInfo.getWorkerCount();
-                Integer defaultSocketReadTimeoutMs = tableInfo.getDefaultSocketReadTimeoutMs();
-                Integer defaultOperationTimeoutMs = tableInfo.getDefaultOperationTimeoutMs();
-
-                Preconditions.checkNotNull(kuduMasters, "kuduMasters could not be null");
-
-                KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMasters);
-                if (null != workerCount) {
-                    kuduClientBuilder.workerCount(workerCount);
-                }
-
-                if (null != defaultOperationTimeoutMs) {
-                    kuduClientBuilder.defaultOperationTimeoutMs(defaultOperationTimeoutMs);
-                }
-                client = kuduClientBuilder.build();
-
+                client = getClient(tableInfo);
                 if (!client.tableExists(tableName)) {
                     throw new IllegalArgumentException("Table Open Failed , please check table exists");
                 }
@@ -243,7 +231,35 @@ private KuduScanner getConn(KuduSideTableInfo tableInfo) {
         }
     }
 
+    private KuduClient getClient(KuduSideTableInfo tableInfo) throws IOException {
+        String kuduMasters = tableInfo.getKuduMasters();
+        Integer workerCount = tableInfo.getWorkerCount();
+        Integer defaultOperationTimeoutMs = tableInfo.getDefaultOperationTimeoutMs();
+
+        Preconditions.checkNotNull(kuduMasters, "kuduMasters could not be null");
+
+        KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMasters);
 
+        if (null != workerCount) {
+            kuduClientBuilder.workerCount(workerCount);
+        }
+
+        if (null != defaultOperationTimeoutMs) {
+            kuduClientBuilder.defaultOperationTimeoutMs(defaultOperationTimeoutMs);
+        }
+
+        if (tableInfo.isEnableKrb()) {
+            UserGroupInformation ugi = KrbUtils.getUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());
+            return ugi.doAs(new PrivilegedAction<KuduClient>() {
+                @Override
+                public KuduClient run() {
+                    return kuduClientBuilder.build();
+                }
+            });
+        } else {
+            return kuduClientBuilder.build();
+        }
+    }
     /**
      * @param builder   创建AsyncKuduScanner对象
      * @param schema    kudu中表约束