code opt

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -25,7 +25,6 @@
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 
 
 /**

core/src/main/java/com/dtstack/flink/sql/option/OptionParser.java

@@ -29,9 +29,7 @@
 import java.util.List;
 import java.util.Map;
 import java.io.File;
-import java.io.FileInputStream;
 import java.net.URLEncoder;
-import java.util.stream.Stream;
 
 import org.apache.commons.codec.Charsets;
 import org.apache.flink.util.FileUtils;

core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java

@@ -22,7 +22,11 @@
 
 import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import com.google.common.collect.Lists;

core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java

@@ -21,7 +21,14 @@
 package com.dtstack.flink.sql.parser;
 
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlMatchRecognize;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.commons.lang3.StringUtils;

core/src/main/java/com/dtstack/flink/sql/side/JoinInfo.java

@@ -25,7 +25,6 @@
 import com.google.common.base.Strings;
 
 import java.io.Serializable;
-import java.util.Map;
 
 /**
  * Join信息

core/src/main/java/com/dtstack/flink/sql/util/DateUtil.java

@@ -23,10 +23,21 @@
 import java.sql.Timestamp;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
-import java.time.*;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeParseException;
-import java.util.*;
+
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Locale;
+import java.util.SimpleTimeZone;
+import java.util.TimeZone;
 import java.util.regex.Pattern;
 
 import static java.time.format.DateTimeFormatter.ISO_INSTANT;
@@ -756,7 +767,8 @@ public static java.sql.Timestamp columnToTimestamp(Object column) {
         if (column == null) {
             return null;
         } else if(column instanceof String) {
-            return null == stringToDate((String) column) ? null : new java.sql.Timestamp(stringToDate((String) column).getTime());
+            Date date = stringToDate((String) column);
+            return null == date ? null : new java.sql.Timestamp(date.getTime());
         } else if (column instanceof Integer) {
             Integer rawData = (Integer) column;
             return new java.sql.Timestamp(rawData.longValue());
@@ -790,7 +802,8 @@ public static Timestamp getTimestampFromStr(String timeStr) {
             Instant instant = Instant.from(ISO_INSTANT.parse(timeStr));
             return new Timestamp(instant.getEpochSecond() * MILLIS_PER_SECOND);
         }
-        return null == stringToDate(timeStr) ? null : new Timestamp(stringToDate(timeStr).getTime());
+        Date date = stringToDate(timeStr);
+        return null == date ? null : new Timestamp(date.getTime());
     }
 
     public static java.sql.Date getDateFromStr(String dateStr) {
@@ -802,7 +815,8 @@ public static java.sql.Date getDateFromStr(String dateStr) {
             Instant instant = Instant.from(ISO_INSTANT.parse(dateStr));
             return new java.sql.Date(instant.toEpochMilli());
         }
-        return null == stringToDate(dateStr) ? null : new java.sql.Date(stringToDate(dateStr).getTime());
+        Date date = stringToDate(dateStr);
+        return null == date ? null : new java.sql.Date(date.getTime());
     }
 
 }

core/src/main/java/com/dtstack/flink/sql/util/MathUtil.java

@@ -23,18 +23,6 @@
 import java.math.BigInteger;
 import java.sql.Date;
 import java.sql.Timestamp;
-import java.text.ParseException;
-import java.text.SimpleDateFormat;
-
-
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalTime;
-import java.time.ZoneOffset;
-import java.util.TimeZone;
-import java.util.regex.Pattern;
-
-import static java.time.format.DateTimeFormatter.ISO_INSTANT;
 
 /**
  * Convert val to specified numeric type

core/src/main/java/com/dtstack/flink/sql/util/ParseUtils.java

@@ -37,16 +37,23 @@
 package com.dtstack.flink.sql.util;
 
 import com.google.common.collect.HashBasedTable;
-import org.apache.calcite.sql.*;
+
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
 
 import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 
-import static org.apache.calcite.sql.SqlKind.*;
+import static org.apache.calcite.sql.SqlKind.AS;
+import static org.apache.calcite.sql.SqlKind.IDENTIFIER;
+import static org.apache.calcite.sql.SqlKind.JOIN;
+
 
 /**
  * @Auther: jiangjunjie

core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java

@@ -29,8 +29,6 @@
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 import java.sql.Timestamp;
-import java.lang.Long;
-
 /**
  * define watermarker
  * Date: 2018/6/29

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java

@@ -32,7 +32,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java

@@ -20,7 +20,10 @@
 
 package com.dtstack.flink.sql.side.hbase;
 
-import com.dtstack.flink.sql.side.*;
+import com.dtstack.flink.sql.side.AbstractSideTableInfo;
+import com.dtstack.flink.sql.side.BaseAllReqRow;
+import com.dtstack.flink.sql.side.FieldInfo;
+import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.hbase.table.HbaseSideTableInfo;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.map.HashedMap;
@@ -34,15 +37,23 @@
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.sql.SQLException;
 import java.sql.Timestamp;
-import java.util.*;
+import java.util.Calendar;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class HbaseAllReqRow extends BaseAllReqRow {

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerKeyedSerializationSchema.java

@@ -4,12 +4,10 @@
 import com.dtstack.flink.sql.format.SerializationMetricWrapper;
 import com.dtstack.flink.sql.sink.kafka.serialization.JsonCRowSerializationSchema;
 import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.formats.json.JsonRowSerializationSchema;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/AvroCRowSerializationSchema.java

@@ -53,7 +53,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.TimeZone;
 import java.util.stream.Collectors;
 
@@ -287,6 +286,7 @@ private Object convertFlinkType(Schema schema, Object object) {
 			case DOUBLE:
 			case BOOLEAN:
 				return object;
+			default:
 		}
 		throw new RuntimeException("Unsupported Avro type:" + schema);
 	}

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/CsvCRowSerializationSchema.java

@@ -47,10 +47,7 @@
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.Arrays;
-import java.util.Iterator;
 import java.util.Objects;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
 
 /**
  * Serialization schema that serializes an object of Flink types into a CSV bytes.
@@ -132,9 +129,9 @@ public Builder setFieldDelimiter(char c) {
 
 		public Builder setLineDelimiter(String delimiter) {
 			Preconditions.checkNotNull(delimiter, "Delimiter must not be null.");
-			if (!delimiter.equals("\n") && !delimiter.equals("\r") && !delimiter.equals("\r\n")) {
+			if (!("\n".equals(delimiter)) && !("\r".equals(delimiter)) && !("\r\n".equals(delimiter))) {
 				throw new IllegalArgumentException(
-					"Unsupported new line delimiter. Only \\n, \\r, or \\r\\n are supported.");
+						"Unsupported new line delimiter. Only \\n, \\r, or \\r\\n are supported.");
 			}
 			this.csvSchema = this.csvSchema.rebuild().setLineSeparator(delimiter).build();
 			return this;

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

@@ -19,7 +19,6 @@
 
 package com.dtstack.flink.sql.source.kafka.table;
 
-import com.dtstack.flink.sql.format.FormatType;
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
 import com.google.common.base.Preconditions;
 

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer09.java

@@ -24,7 +24,6 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer09;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer09Factory.java

@@ -23,7 +23,6 @@
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 
 import java.util.Optional;
 import java.util.Properties;

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

@@ -21,18 +21,15 @@
 
 import com.dtstack.flink.sql.source.kafka.table.KafkaSourceTableInfo;
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
-import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.types.Row;
 
-import java.util.Map;
 import java.util.Properties;
 
 /**

kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer010Factory.java

@@ -23,7 +23,6 @@
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 
 import java.util.Optional;
 import java.util.Properties;

kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer011Factory.java

@@ -23,7 +23,6 @@
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.types.Row;
 
 import java.util.Optional;
 import java.util.Properties;

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

@@ -20,9 +20,7 @@
 
 package com.dtstack.flink.sql.launcher;
 
-import com.aiweiergou.tool.logger.api.ChangeLogLevelProcess;
 import com.dtstack.flink.sql.constrant.ConfigConstrant;
-import com.dtstack.flink.sql.launcher.perjob.PerJobClusterClientBuilder;
 import com.google.common.collect.Lists;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.TypeReference;
@@ -43,14 +41,9 @@
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.util.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-import java.io.BufferedReader;
 import java.io.File;
-import java.io.FileInputStream;
 import java.io.IOException;
-import java.io.InputStreamReader;
 import java.net.URLDecoder;
 import java.util.LinkedList;
 import java.util.List;

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java

@@ -32,14 +32,12 @@
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.ArrayList;

mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java

@@ -37,8 +37,6 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;

mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java

@@ -44,13 +44,6 @@
 import com.mongodb.async.client.MongoClients;
 import com.mongodb.async.client.MongoCollection;
 import com.mongodb.async.client.MongoDatabase;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import com.google.common.collect.Lists;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.functions.async.ResultFuture;
-import org.apache.flink.table.runtime.types.CRow;
-import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
-import org.apache.flink.types.Row;
 import org.bson.Document;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java

@@ -35,8 +35,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 
 /**
  * Reason: