merge 1.8

Author: todd5167

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleSink.java

@@ -73,6 +73,7 @@ public TypeInformation<?>[] getFieldTypes() {
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         // flink 1.9 use consumeDataStream
+        consumeDataStream(dataStream);
     }
 
     @Override

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -60,6 +60,7 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
@@ -149,7 +150,7 @@ public static boolean checkRemoteSqlPluginPath(String remoteSqlPluginPath, Strin
 
     public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInfo) throws Exception {
         StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExeEnv(paramsInfo.getConfProp(), paramsInfo.getDeployMode());
-        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
+        StreamTableEnvironment tableEnv = getStreamTableEnv(env,paramsInfo.getConfProp());
 
         SqlParser.setLocalSqlPluginRoot(paramsInfo.getLocalSqlPluginPath());
         SqlTree sqlTree = SqlParser.parseSql(paramsInfo.getSql());
@@ -284,11 +285,7 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
                 Table adaptTable = adaptSql == null ? table : tableEnv.sqlQuery(adaptSql);
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getFieldTypes(), adaptTable.getSchema().getFieldNames());
-                DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
-                        .map((Tuple2<Boolean, Row> f0) -> {
-                            return f0.f1;
-                        })
-                        .returns(typeInfo);
+                DataStream adaptStream = tableEnv.toAppendStream(adaptTable, typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());
 
@@ -352,6 +349,19 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         return env;
     }
 
+    private static StreamTableEnvironment getStreamTableEnv(StreamExecutionEnvironment env, Properties confProperties) {
+        // use blink and streammode
+        EnvironmentSettings settings = EnvironmentSettings.newInstance()
+                .useBlinkPlanner()
+                .inStreamingMode()
+                .build();
+
+        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
+        StreamEnvConfigManager.streamTableEnvironmentStateTTLConfig(tableEnv, confProperties);
+        return tableEnv;
+    }
+
+
     public static void setLogLevel(ParamsInfo paramsInfo){
         String logLevel = paramsInfo.getConfProp().getProperty(ConfigConstrant.LOG_LEVEL_KEY);
         if(org.apache.commons.lang3.StringUtils.isBlank(logLevel)){

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

@@ -117,6 +117,10 @@ protected void beforeDeserialize() throws IOException {
     protected void afterDeserialize() throws IOException {
     }
 
+    public DeserializationSchema<Row> getDeserializationSchema() {
+        return deserializationSchema;
+    }
+
     public RuntimeContext getRuntimeContext() {
         return runtimeContext;
     }

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -58,8 +58,10 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
     private final String[] fieldNames;
     private final TypeInformation<?>[] fieldTypes;
     private List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
+    private TypeInformation<Row> typeInfo;
 
     public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos) {
+        this.typeInfo = typeInfo;
         this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
         this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
         this.rowAndFieldMapping = rowAndFieldMapping;
@@ -178,5 +180,8 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         }
     }
 
-
+    @Override
+    public TypeInformation<Row> getProducedType() {
+        return typeInfo;
+    }
 }

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java

@@ -120,11 +120,17 @@ private RichSinkFunction createEsSinkFunction() {
 
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
+        consumeDataStream(dataStream);
+    }
+
+    @Override
+    public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         RichSinkFunction richSinkFunction = createEsSinkFunction();
         DataStreamSink streamSink = dataStream.addSink(richSinkFunction);
         if (parallelism > 0) {
             streamSink.setParallelism(parallelism);
         }
+        return streamSink;
     }
 
     @Override

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

@@ -19,7 +19,6 @@
 
 import com.dtstack.flink.sql.format.FormatType;
 import com.dtstack.flink.sql.format.SerializationMetricWrapper;
-import com.dtstack.flink.sql.format.dtnest.DtNestRowDeserializationSchema;
 import com.dtstack.flink.sql.sink.kafka.table.KafkaSinkTableInfo;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.serialization.SerializationSchema;