Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.8.0_dev

todd5167 · todd5167 · commit 5f25a8581429 · 2019-09-25T22:25:19.000+08:00
diff --git a/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -64,6 +64,9 @@ public class KafkaSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 	/** Partitioner to select Kafka partition for each item. */
 	protected Optional<FlinkKafkaPartitioner<Row>> partitioner;
 
+	protected int parallelism;
+
+
 
 	@Override
 	public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
@@ -85,11 +88,16 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
 		this.fieldTypes = types;
 
 		TableSchema.Builder schemaBuilder = TableSchema.builder();
-		for (int i=0;i<fieldNames.length;i++){
-			schemaBuilder.field(fieldNames[i], fieldTypes[i]);
-		}
+		for (int i=0;i<fieldNames.length;i++) {
+            schemaBuilder.field(fieldNames[i], fieldTypes[i]);
+        }
 		this.schema = schemaBuilder.build();
 
+		Integer parallelism = kafka09SinkTableInfo.getParallelism();
+		if (parallelism != null) {
+			this.parallelism = parallelism;
+		}
+
 		this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType().getTypeAt(1));
 		return this;
 	}
@@ -111,7 +119,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
 		DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
 			return record.f1;
-		}).returns(getOutputType().getTypeAt(1));
+		}).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
 
 		kafkaTableSink.emitDataStream(ds);
 	}
diff --git a/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java b/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java
@@ -42,6 +42,10 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
 
         kafka09SinkTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSinkTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka09SinkTableInfo.setTopic(MathUtil.getString(props.get(KafkaSinkTableInfo.TOPIC_KEY.toLowerCase())));
+
+        Integer parallelism = MathUtil.getIntegerVal(props.get(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase()));
+        kafka09SinkTableInfo.setParallelism(parallelism);
+
         for (String key : props.keySet()) {
             if (!key.isEmpty() && key.startsWith("kafka.")) {
                 kafka09SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -29,6 +29,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+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;
@@ -114,6 +115,12 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
         String fields = StringUtils.join(kafka09SourceTableInfo.getFields(), ",");
         String sourceOperatorName = SOURCE_OPERATOR_NAME_TPL.replace("${topic}", topicName).replace("${table}", sourceTableInfo.getName());
-        return tableEnv.fromDataStream(env.addSource(kafkaSrc, sourceOperatorName, typeInformation), fields);
+
+        DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
+        Integer parallelism = kafka09SourceTableInfo.getParallelism();
+        if (parallelism != null) {
+            kafkaSource.setParallelism(parallelism);
+        }
+        return tableEnv.fromDataStream(kafkaSource, fields);
     }
 }
diff --git a/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -58,6 +58,8 @@ public class KafkaSink implements RetractStreamTableSink<Row>, IStreamSinkGener<
 
     protected Properties properties;
 
+    protected int parallelism;
+
     /** Serialization schema for encoding records to Kafka. */
     protected SerializationSchema serializationSchema;
 
@@ -87,12 +89,18 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         }
         this.fieldTypes = types;
 
+
         TableSchema.Builder schemaBuilder = TableSchema.builder();
         for (int i=0;i<fieldNames.length;i++){
             schemaBuilder.field(fieldNames[i], fieldTypes[i]);
         }
         this.schema = schemaBuilder.build();
 
+        Integer parallelism = kafka10SinkTableInfo.getParallelism();
+        if (parallelism != null) {
+            this.parallelism = parallelism;
+        }
+
         this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType().getTypeAt(1));
         return this;
     }
@@ -114,7 +122,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
         DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
             return record.f1;
-        }).returns(getOutputType().getTypeAt(1));
+        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }
diff --git a/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java b/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java
@@ -42,6 +42,10 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
 
         kafka10SinkTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSinkTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka10SinkTableInfo.setTopic(MathUtil.getString(props.get(KafkaSinkTableInfo.TOPIC_KEY.toLowerCase())));
+
+        Integer parallelism = MathUtil.getIntegerVal(props.get(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase()));
+        kafka10SinkTableInfo.setParallelism(parallelism);
+
         for (String key : props.keySet()) {
             if (!key.isEmpty() && key.startsWith("kafka.")) {
                 kafka10SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -28,6 +28,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
@@ -116,6 +117,12 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
 		String fields = StringUtils.join(kafka010SourceTableInfo.getFields(), ",");
 		String sourceOperatorName = SOURCE_OPERATOR_NAME_TPL.replace("${topic}", topicName).replace("${table}", sourceTableInfo.getName());
-		return tableEnv.fromDataStream(env.addSource(kafkaSrc, sourceOperatorName, typeInformation), fields);
+
+		DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
+		Integer parallelism = kafka010SourceTableInfo.getParallelism();
+		if (parallelism != null) {
+			kafkaSource.setParallelism(parallelism);
+		}
+		return tableEnv.fromDataStream(kafkaSource, fields);
 	}
 }
diff --git a/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -56,6 +56,8 @@ public class KafkaSink  implements RetractStreamTableSink<Row>, IStreamSinkGener
 
     protected String topic;
 
+    protected int parallelism;
+
     protected Properties properties;
 
     /** Serialization schema for encoding records to Kafka. */
@@ -88,10 +90,16 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.fieldTypes = types;
 
         TableSchema.Builder schemaBuilder = TableSchema.builder();
-        for (int i=0;i<fieldNames.length;i++){
+        for (int i=0;i<fieldNames.length;i++) {
             schemaBuilder.field(fieldNames[i], fieldTypes[i]);
         }
         this.schema = schemaBuilder.build();
+
+        Integer parallelism = kafka11SinkTableInfo.getParallelism();
+        if (parallelism != null) {
+            this.parallelism = parallelism;
+        }
+
         this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType().getTypeAt(1));
         return this;
     }
@@ -110,9 +118,10 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 partitioner,
                 serializationSchema
         );
+
         DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
             return record.f1;
-        }).returns(getOutputType().getTypeAt(1));
+        }).returns(getOutputType().getTypeAt(1)).setParallelism(parallelism);
 
         kafkaTableSink.emitDataStream(ds);
     }
diff --git a/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java b/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java
@@ -50,6 +50,10 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
 
         kafka11SinkTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSinkTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka11SinkTableInfo.setTopic(MathUtil.getString(props.get(KafkaSinkTableInfo.TOPIC_KEY.toLowerCase())));
+
+        Integer parallelism = MathUtil.getIntegerVal(props.get(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase()));
+        kafka11SinkTableInfo.setParallelism(parallelism);
+
         for (String key : props.keySet()) {
             if (!key.isEmpty() && key.startsWith("kafka.")) {
                 kafka11SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -28,6 +28,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
@@ -116,6 +117,12 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
 		String fields = StringUtils.join(kafka011SourceTableInfo.getFields(), ",");
 		String sourceOperatorName = SOURCE_OPERATOR_NAME_TPL.replace("${topic}", topicName).replace("${table}", sourceTableInfo.getName());
-		return tableEnv.fromDataStream(env.addSource(kafkaSrc, sourceOperatorName, typeInformation), fields);
+
+		DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
+		Integer parallelism = kafka011SourceTableInfo.getParallelism();
+		if (parallelism != null) {
+			kafkaSource.setParallelism(parallelism);
+		}
+		return tableEnv.fromDataStream(kafkaSource, fields);
 	}
 }
diff --git a/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java b/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java
@@ -69,11 +69,12 @@ public void open(Configuration parameters) throws Exception {
                 .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
                 .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
 
-
+        System.setProperty("vertx.disableFileCPResolving", "true");
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
         vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
     }
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java
@@ -59,11 +59,14 @@ public void open(Configuration parameters) throws Exception {
                 .put("provider_class", DT_PROVIDER_CLASS)
                 .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
                 .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
-                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);;
+                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
+
+        System.setProperty("vertx.disableFileCPResolving", "true");
 
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
         vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, oracleClientConfig));
     }
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/RdbSink.java
@@ -69,6 +69,8 @@ public abstract class RdbSink implements RetractStreamTableSink<Row>, Serializab
 
     protected String tableName;
 
+    protected String registerTabName;
+
     protected String sql;
 
     protected List<String> primaryKeys;
@@ -112,6 +114,7 @@ public RdbSink genStreamSink(TargetTableInfo targetTableInfo) {
         String tmpUserName = rdbTableInfo.getUserName();
         String tmpPassword = rdbTableInfo.getPassword();
         String tmpTableName = rdbTableInfo.getTableName();
+        String tmpRegisterName = rdbTableInfo.getName();
 
         Integer tmpSqlBatchSize = rdbTableInfo.getBatchSize();
         if (tmpSqlBatchSize != null) {
@@ -136,6 +139,7 @@ public RdbSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.userName = tmpUserName;
         this.password = tmpPassword;
         this.tableName = tmpTableName;
+        this.registerTabName = tmpRegisterName;
         this.primaryKeys = rdbTableInfo.getPrimaryKeys();
         this.dbType = rdbTableInfo.getType();
 
@@ -193,7 +197,7 @@ protected void buildSqlTypes(List<Class> fieldTypeArray) {
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
         RichSinkFunction richSinkFunction = createJdbcSinkFunc();
         DataStreamSink streamSink = dataStream.addSink(richSinkFunction);
-        streamSink.name(tableName);
+        streamSink.name(registerTabName);
         if (parallelism > 0) {
             streamSink.setParallelism(parallelism);
         }
diff --git a/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java b/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java
@@ -59,9 +59,13 @@ public void open(Configuration parameters) throws Exception {
                 .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
                 .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
 
+
+        System.setProperty("vertx.disableFileCPResolving", "true");
+
         VertxOptions vo = new VertxOptions();
         vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
         vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSQLClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));
     }
