Merge branch '1.10_release_4.0.x' into 'hotfix_1.10_4.0.x_31804'

FlechazoW · FlechazoW · commit 40a86e7439f2 · 2020-10-29T14:20:12.000+08:00
# Conflicts:
#   localTest/pom.xml
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
@@ -73,7 +73,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
@@ -72,7 +72,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
@@ -52,6 +52,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -76,7 +77,7 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     // cast(value as string) -> cast('value' as string)  cast(value as timestamp) -> cast('value' as timestamp)
     private static final Pattern TYPE_PATTERN = Pattern.compile("cast\\((.*) as (.*)\\)");
     //specific type which values need to be quoted
-    private static final String[] NEED_QUOTE_TYPE = {"string", "timestamp"};
+    private static final String[] NEED_QUOTE_TYPE = {"string", "timestamp", "varchar"};
 
     private static final Integer DEFAULT_CONN_TIME_OUT = 60;
     private static final int RECEIVE_DATA_PRINT_FREQUENCY = 1000;
@@ -85,8 +86,6 @@ public class ImpalaOutputFormat extends AbstractDtRichOutputFormat<Tuple2<Boolea
     private static final String KUDU_TYPE = "kudu";
     private static final String UPDATE_MODE = "update";
     private static final String PARTITION_CONSTANT = "PARTITION";
-    private static final String STRING_TYPE = "STRING";
-    private static final String TIMESTAMP_TYPE = "TIMESTAMP";
     private static final String DRIVER_NAME = "com.cloudera.impala.jdbc41.Driver";
 
     private static final String VALUES_CONDITION = "${valuesCondition}";
@@ -208,7 +207,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();
@@ -363,24 +362,28 @@ private List<String> rebuildFieldNameListAndTypeList(List<String> fieldNames, Li
      * @return quoted condition
      */
     private String valueConditionAddQuotation(String valueCondition) {
-        final String[] valueConditionCopy = {valueCondition};
         String[] temps = valueCondition.split(",");
+        List<String> replacedItem = new ArrayList<>();
         Arrays.stream(temps).forEach(
                 item -> {
                     Matcher matcher = TYPE_PATTERN.matcher(item);
                     while (matcher.find()) {
                         String value = matcher.group(1);
                         String type = matcher.group(2);
 
-                        if (Arrays.asList(NEED_QUOTE_TYPE).contains(type)) {
-                            if (!"null".equals(value)) {
-                                valueConditionCopy[0] = valueConditionCopy[0].replace(value, "'" + value + "'");
+                        for (String needQuoteType : NEED_QUOTE_TYPE) {
+                            if (type.contains(needQuoteType)) {
+                                if (!"null".equals(value)) {
+                                    item = item.replace(value, "'" + value + "'");
+                                }
                             }
                         }
                     }
+                    replacedItem.add(item);
                 }
         );
-        return "(" + valueConditionCopy[0] + ")";
+
+        return "(" + String.join(", ", replacedItem) + ")";
     }
 
     @Override
@@ -580,8 +583,10 @@ private String buildTableFieldsCondition(List<String> fieldNames, String partiti
     private String buildValuesCondition(List<String> fieldTypes, Row row) {
         String valuesCondition = fieldTypes.stream().map(
                 f -> {
-                    if (Arrays.asList(NEED_QUOTE_TYPE).contains(f.toLowerCase())) {
-                        return String.format("cast(? as %s)", f.toLowerCase());
+                    for(String item : NEED_QUOTE_TYPE) {
+                        if (f.toLowerCase().contains(item)) {
+                            return String.format("cast(? as %s)", f.toLowerCase());
+                        }
                     }
                     return "?";
                 }).collect(Collectors.joining(", "));
@@ -745,6 +750,8 @@ public ImpalaOutputFormat build() {
                 checkNotNull(format.password, "password is required!");
             }
 
+            checkNotNull(format.storeType, "storeType is required!");
+
             return format;
         }
 
diff --git a/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaSink.java b/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaSink.java
@@ -46,7 +46,6 @@
 public class ImpalaSink implements RetractStreamTableSink<Row>, IStreamSinkGener<ImpalaSink> {
 
     private static final String DEFAULT_STORE_TYPE = "kudu";
-    private static final String DEFAULT_PARTITION_MODE = "dynamic";
 
     protected String[] fieldNames;
     TypeInformation<?>[] fieldTypes;
diff --git a/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/table/ImpalaSinkParser.java b/impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/table/ImpalaSinkParser.java
@@ -25,6 +25,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Date: 2020/10/14
@@ -76,6 +77,8 @@ public class ImpalaSinkParser extends AbstractTableParser {
 
     private static final String KUDU_TYPE = "kudu";
 
+    private static final String DEFAULT_STORE_TYPE = "kudu";
+
     private static final String STORE_TYPE_KEY = "storeType";
 
     private static final String KRB_DEFAULT_REALM = "HADOOP.COM";
@@ -123,13 +126,13 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         }
 
         String storeType = MathUtil.getString(props.get(STORE_TYPE_KEY.toLowerCase()));
-        impalaTableInfo.setStoreType(storeType);
+        impalaTableInfo.setStoreType(Objects.isNull(storeType) ? DEFAULT_STORE_TYPE : storeType);
 
         String enablePartitionStr = (String) props.get(ENABLE_PARTITION_KEY.toLowerCase());
         boolean enablePartition = MathUtil.getBoolean(enablePartitionStr == null ? "false" : enablePartitionStr);
         impalaTableInfo.setEnablePartition(enablePartition);
 
-        if (!storeType.equalsIgnoreCase(KUDU_TYPE) && enablePartition) {
+        if (!impalaTableInfo.getStoreType().equalsIgnoreCase(KUDU_TYPE) && enablePartition) {
             String partitionFields = MathUtil.getString(props.get(PARTITION_FIELDS_KEY.toLowerCase()));
             impalaTableInfo.setPartitionFields(partitionFields);
         }
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -26,6 +26,7 @@
 import com.dtstack.flink.sql.util.MathUtil;
 
 import java.util.Map;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 /**
@@ -46,7 +47,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setType(MathUtil.getString(props.get(KafkaSourceTableInfo.TYPE_KEY.toLowerCase())));
         kafkaSourceTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(KafkaSourceTableInfo.PARALLELISM_KEY.toLowerCase())));
         kafkaSourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
-        kafkaSourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
+        kafkaSourceTableInfo.setGroupId(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase(), UUID.randomUUID().toString().replace("-", ""))));
         kafkaSourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase()), false));
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase(), EKafkaOffset.LATEST.name().toLowerCase())));
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
@@ -227,7 +227,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
@@ -132,7 +132,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
`@@ -227,7 +227,7 @@ private KuduClient getClient(KuduSideTableInfo tableInfo) throws IOException {`
`227`	`227`	`}`
`228`	`228`
`229`	`229`	`if (tableInfo.isEnableKrb()) {`
`230`		`- UserGroupInformation ugi = KrbUtils.getUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());`
	`230`	`+ UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(tableInfo.getPrincipal(), tableInfo.getKeytab(), tableInfo.getKrb5conf());`
`231`	`231`	`return ugi.doAs(new PrivilegedAction<KuduClient>() {`
`232`	`232`	`@Override`
`233`	`233`	`public KuduClient run() {`
Original file line number	Diff line number	Diff line change
`@@ -132,7 +132,7 @@ private AsyncKuduClient getClient() throws IOException {`
`132`	`132`	`}`
`133`	`133`
`134`	`134`	`if (kuduSideTableInfo.isEnableKrb()) {`
`135`		`- UserGroupInformation ugi = KrbUtils.getUgi(`
	`135`	`+ UserGroupInformation ugi = KrbUtils.loginAndReturnUgi(`
`136`	`136`	`kuduSideTableInfo.getPrincipal(),`
`137`	`137`	`kuduSideTableInfo.getKeytab(),`
`138`	`138`	`kuduSideTableInfo.getKrb5conf()`