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

Author: a49a

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -18,7 +18,18 @@
 
 package com.dtstack.flink.sql.side.cassandra;
 
-import com.datastax.driver.core.*;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
+import com.datastax.driver.core.QueryOptions;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.SocketOptions;
 import com.datastax.driver.core.policies.DowngradingConsistencyRetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,10 +43,7 @@
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Collector;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -72,27 +80,6 @@ public CassandraAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<Field
         super(new com.dtstack.flink.sql.side.cassandra.CassandraAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
-    @Override
-    public Row fillData(Row input, Object sideInput) {
-        Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
-        Row row = new Row(sideInfo.getOutFieldInfoList().size());
-        for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
-            Object obj = input.getField(entry.getValue());
-            obj = convertTimeIndictorTypeInfo(entry.getValue(), obj);
-            row.setField(entry.getKey(), obj);
-        }
-
-        for (Map.Entry<Integer, String> entry : sideInfo.getSideFieldNameIndex().entrySet()) {
-            if (cacheInfo == null) {
-                row.setField(entry.getKey(), null);
-            } else {
-                row.setField(entry.getKey(), cacheInfo.get(entry.getValue()));
-            }
-        }
-
-        return row;
-    }
-
     @Override
     protected void initCache() throws SQLException {
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();

cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java

@@ -72,57 +72,4 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
         LOG.info("---------side_exe_sql-----\n{}" + sqlCondition);
     }
 
-
-    @Override
-    public void dealOneEqualCon(SqlNode sqlNode, String sideTableName) {
-        if (sqlNode.getKind() != SqlKind.EQUALS) {
-            throw new RuntimeException("not equal operator.");
-        }
-
-        SqlIdentifier left = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[0];
-        SqlIdentifier right = (SqlIdentifier) ((SqlBasicCall) sqlNode).getOperands()[1];
-
-        String leftTableName = left.getComponent(0).getSimple();
-        String leftField = left.getComponent(1).getSimple();
-
-        String rightTableName = right.getComponent(0).getSimple();
-        String rightField = right.getComponent(1).getSimple();
-
-        if (leftTableName.equalsIgnoreCase(sideTableName)) {
-            equalFieldList.add(leftField);
-            int equalFieldIndex = -1;
-            for (int i = 0; i < getFieldNames().length; i++) {
-                String fieldName = getFieldNames()[i];
-                if (fieldName.equalsIgnoreCase(rightField)) {
-                    equalFieldIndex = i;
-                }
-            }
-            if (equalFieldIndex == -1) {
-                throw new RuntimeException("can't deal equal field: " + sqlNode);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        } else if (rightTableName.equalsIgnoreCase(sideTableName)) {
-
-            equalFieldList.add(rightField);
-            int equalFieldIndex = -1;
-            for (int i = 0; i < getFieldNames().length; i++) {
-                String fieldName = getFieldNames()[i];
-                if (fieldName.equalsIgnoreCase(leftField)) {
-                    equalFieldIndex = i;
-                }
-            }
-            if (equalFieldIndex == -1) {
-                throw new RuntimeException("can't deal equal field: " + sqlNode.toString());
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        } else {
-            throw new RuntimeException("resolve equalFieldList error:" + sqlNode.toString());
-        }
-
-    }
-
 }

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/side/BaseAllReqRow.java

@@ -23,6 +23,7 @@
 import com.dtstack.flink.sql.util.RowDataComplete;
 import org.apache.calcite.sql.JoinType;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
@@ -34,6 +35,8 @@
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.time.LocalDateTime;
+import java.util.Map;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -42,7 +45,6 @@
  * Reason:
  * Date: 2018/9/18
  * Company: www.dtstack.com
- *
  * @author xuchao
  */
 
@@ -52,6 +54,8 @@ public abstract class BaseAllReqRow extends RichFlatMapFunction<Row, BaseRow> im
 
     public static final long LOAD_DATA_ERROR_SLEEP_TIME = 5_000L;
 
+    public static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     protected BaseSideInfo sideInfo;
 
     private ScheduledExecutorService es;
@@ -95,6 +99,45 @@ protected void sendOutputRow(Row value, Object sideInput, Collector<BaseRow> out
         RowDataComplete.collectRow(out, row);
     }
 
+    @Override
+    public Row fillData(Row input, Object sideInput) {
+        Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
+        Row row = new Row(sideInfo.getOutFieldInfoList().size());
+
+        for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
+            // origin value
+            Object obj = input.getField(entry.getValue());
+            obj = dealTimeAttributeType(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass(), obj);
+            row.setField(entry.getKey(), obj);
+        }
+
+        for (Map.Entry<Integer, String> entry : sideInfo.getSideFieldNameIndex().entrySet()) {
+            if (cacheInfo == null) {
+                row.setField(entry.getKey(), null);
+            } else {
+                row.setField(entry.getKey(), cacheInfo.get(entry.getValue()));
+            }
+        }
+        return row;
+    }
+
+    /**
+     * covert flink time attribute.Type information for indicating event or processing time.
+     * However, it behaves like a regular SQL timestamp but is serialized as Long.
+     *
+     * @param entry
+     * @param obj
+     * @return
+     */
+    protected Object dealTimeAttributeType(Class<? extends TypeInformation> entry, Object obj) {
+        boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(entry);
+        if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
+        }
+        return obj;
+    }
+
     @Override
     public void close() throws Exception {
         if (null != es && !es.isShutdown()) {