Merge remote-tracking branch 'git_flinkStreamSQL/1.10_release' into 1.10_release

FlechazoW · FlechazoW · commit e0a7da915a0d · 2020-10-23T11:58:58.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/BaseSideInfo.java b/core/src/main/java/com/dtstack/flink/sql/side/BaseSideInfo.java
@@ -75,12 +75,15 @@ public abstract class BaseSideInfo implements Serializable{
 
     protected AbstractSideCache sideCache;
 
+    protected JoinInfo joinInfo;
+
     public BaseSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
                         AbstractSideTableInfo sideTableInfo){
         this.rowTypeInfo = rowTypeInfo;
         this.outFieldInfoList = outFieldInfoList;
         this.joinType = joinInfo.getJoinType();
         this.sideTableInfo = sideTableInfo;
+        this.joinInfo = joinInfo;
         parseSelectFields(joinInfo);
         buildEqualInfo(joinInfo, sideTableInfo);
     }
@@ -178,9 +181,7 @@ private void evalEquation(SqlIdentifier left, SqlIdentifier right, String sideTa
      */
     private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier) {
         String tableName = identifier.getComponent(0).getSimple();
-        String sideTableName = sideTableInfo.getName();
-        String errorMsg = "only support set side table constant field, error field " + identifier;
-        Preconditions.checkState(tableName.equals(sideTableName), errorMsg);
+        checkSupport(identifier);
         String fieldName = identifier.getComponent(1).getSimple();
         Object constant = literal.getValue();
         List<PredicateInfo> predicateInfos = sideTableInfo.getPredicateInfoes();
@@ -194,6 +195,22 @@ private void evalConstantEquation(SqlLiteral literal, SqlIdentifier identifier)
         predicateInfos.add(predicate);
     }
 
+    private void checkSupport(SqlIdentifier identifier) {
+        String tableName = identifier.getComponent(0).getSimple();
+        String sideTableName;
+        String sideTableAlias;
+        if (joinInfo.isLeftIsSideTable()) {
+            sideTableName = joinInfo.getLeftTableName();
+            sideTableAlias = joinInfo.getLeftTableAlias();
+        } else {
+            sideTableName = joinInfo.getRightTableName();
+            sideTableAlias = joinInfo.getRightTableAlias();
+        }
+        boolean isSide = tableName.equals(sideTableName) || tableName.equals(sideTableAlias);
+        String errorMsg = "only support set side table constant field, error field " + identifier;
+        Preconditions.checkState(isSide, errorMsg);
+    }
+
     private void associateField(String sourceTableField, String sideTableField, SqlNode sqlNode) {
         String errorMsg = "can't deal equal field: " + sqlNode;
         equalFieldList.add(sideTableField);
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/JoinInfo.java b/core/src/main/java/com/dtstack/flink/sql/side/JoinInfo.java
@@ -55,15 +55,15 @@ public class JoinInfo implements Serializable {
 
     private String rightTableAlias;
 
-    private SqlNode leftNode;
+    private transient SqlNode leftNode;
 
-    private SqlNode rightNode;
+    private transient SqlNode rightNode;
 
-    private SqlNode condition;
+    private transient SqlNode condition;
 
-    private SqlNode selectFields;
+    private transient SqlNode selectFields;
 
-    private SqlNode selectNode;
+    private transient SqlNode selectNode;
 
     private JoinType joinType;
 
diff --git a/core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java b/core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java
@@ -20,8 +20,11 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authentication.util.KerberosName;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import sun.security.krb5.Config;
+import sun.security.krb5.KrbException;
 
 import java.io.IOException;
 
@@ -40,9 +43,16 @@ public class KrbUtils {
 //    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 {
+    public static UserGroupInformation loginAndReturnUgi(String principal, String keytabPath, String krb5confPath) throws IOException {
         LOG.info("Kerberos login with principal: {} and keytab: {}", principal, keytabPath);
         System.setProperty(KRB5_CONF_KEY, krb5confPath);
+        // 不刷新会读/etc/krb5.conf
+        try {
+            Config.refresh();
+            KerberosName.resetDefaultRealm();
+        } catch (KrbException e) {
+            LOG.warn("resetting default realm failed, current default realm will still be used.", e);
+        }
         // TODO 尚未探索出此选项的意义，以后研究明白方可打开
 //        System.setProperty(SUBJECT_ONLY_KEY, FALSE_STR);
         Configuration configuration = new Configuration();
diff --git a/core/src/test/java/com/dtstack/flink/sql/util/KrbUtilsTest.java b/core/src/test/java/com/dtstack/flink/sql/util/KrbUtilsTest.java
@@ -30,12 +30,12 @@
  **/
 public class KrbUtilsTest {
     @Test
-    public void testGetUgi() throws IOException {
+    public void testLoginAndReturnUgi() throws IOException {
         String principal = "";
         String keytabPath = "";
         String krb5confPath = "";
         try {
-            KrbUtils.getUgi(principal, keytabPath, krb5confPath);
+            KrbUtils.loginAndReturnUgi(principal, keytabPath, krb5confPath);
         } catch (IllegalArgumentException e) {
             Assert.assertEquals(e.getMessage(), "Can't get Kerberos realm");
         }
diff --git a/impala/impala-side/impala-all-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAllReqRow.java b/impala/impala-side/impala-all-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAllReqRow.java
@@ -71,7 +71,7 @@ public Connection getConn(String dbUrl, String userName, String password) {
                 String keyTabFilePath = impalaSideTableInfo.getKeyTabFilePath();
                 String krb5FilePath = impalaSideTableInfo.getKrb5FilePath();
                 String principal = impalaSideTableInfo.getPrincipal();
-                UserGroupInformation ugi = KrbUtils.getUgi(principal, keyTabFilePath, krb5FilePath);
+                UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(principal, keyTabFilePath, krb5FilePath);
                 connection = ugi.doAs(new PrivilegedExceptionAction<Connection>() {
                     @Override
                     public Connection run() throws SQLException {
diff --git a/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java b/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java
@@ -70,7 +70,7 @@ public void open(Configuration parameters) throws Exception {
             String keyTabFilePath = impalaSideTableInfo.getKeyTabFilePath();
             String krb5FilePath = impalaSideTableInfo.getKrb5FilePath();
             String principal = impalaSideTableInfo.getPrincipal();
-            ugi = KrbUtils.getUgi(principal, keyTabFilePath, krb5FilePath);
+            ugi = KrbUtils.loginAndReturnUgi(principal, keyTabFilePath, krb5FilePath);
             openJdbc(parameters);
         } else {
             openJdbc(parameters);
diff --git a/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java b/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java
@@ -208,7 +208,7 @@ private void initScheduledTask(Long batchWaitInterval) {
 
     private void openConnect() throws IOException {
         if (authMech == 1) {
-            UserGroupInformation ugi = KrbUtils.getUgi(principal, keytabPath, krb5confPath);
+            UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(principal, keytabPath, krb5confPath);
             try {
                 ugi.doAs((PrivilegedExceptionAction<Void>) () -> {
                     openJdbc();
diff --git a/kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java b/kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java
@@ -210,7 +210,7 @@ private KuduClient getClient(KuduSideTableInfo tableInfo) throws IOException {
         }
 
         if (tableInfo.isEnableKrb()) {
-            UserGroupInformation ugi = KrbUtils.getUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());
+            UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());
             return ugi.doAs(new PrivilegedAction<KuduClient>() {
                 @Override
                 public KuduClient run() {
diff --git a/kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java b/kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java
@@ -119,7 +119,7 @@ private AsyncKuduClient getClient() throws IOException {
         }
 
         if (kuduSideTableInfo.isEnableKrb()) {
-            UserGroupInformation ugi = KrbUtils.getUgi(
+            UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(
                 kuduSideTableInfo.getPrincipal(),
                 kuduSideTableInfo.getKeytab(),
                 kuduSideTableInfo.getKrb5conf()
diff --git a/kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java b/kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java
@@ -116,7 +116,7 @@ private void establishConnection() throws IOException {
         }
 
         if (enableKrb) {
-            UserGroupInformation ugi = KrbUtils.getUgi(
+            UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(
                 principal,
                 keytab,
                 krb5conf

Original file line number	Diff line number	Diff line change
`@@ -210,7 +210,7 @@ private KuduClient getClient(KuduSideTableInfo tableInfo) throws IOException {`
`210`	`210`	`}`
`211`	`211`
`212`	`212`	`if (tableInfo.isEnableKrb()) {`
`213`		`- UserGroupInformation ugi = KrbUtils.getUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());`
	`213`	`+ UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());`
`214`	`214`	`return ugi.doAs(new PrivilegedAction<KuduClient>() {`
`215`	`215`	`@Override`
`216`	`216`	`public KuduClient run() {`
Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ private AsyncKuduClient getClient() throws IOException {`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`if (kuduSideTableInfo.isEnableKrb()) {`
`122`		`- UserGroupInformation ugi = KrbUtils.getUgi(`
	`122`	`+ UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(`
`123`	`123`	`kuduSideTableInfo.getPrincipal(),`
`124`	`124`	`kuduSideTableInfo.getKeytab(),`
`125`	`125`	`kuduSideTableInfo.getKrb5conf()`
Original file line number	Diff line number	Diff line change
`@@ -116,7 +116,7 @@ private void establishConnection() throws IOException {`
`116`	`116`	`}`
`117`	`117`
`118`	`118`	`if (enableKrb) {`
`119`		`- UserGroupInformation ugi = KrbUtils.getUgi(`
	`119`	`+ UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(`
`120`	`120`	`principal,`
`121`	`121`	`keytab,`
`122`	`122`	`krb5conf`