support kafka parameter

Author: todd5167

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

@@ -64,6 +64,9 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
 		this.fieldTypes = types;
 
 		properties = new Properties();
+		for (String key:kafka09SinkTableInfo.getKafkaParamKeys()) {
+			properties.setProperty(key, kafka09SinkTableInfo.getKafkaParam(key));
+		}
 		properties.setProperty("bootstrap.servers", kafka09SinkTableInfo.getBootstrapServers());
 
 		this.serializationSchema = new JsonRowSerializationSchema(getOutputType());

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

@@ -42,6 +42,11 @@ 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())));
+        for (String key:props.keySet()) {
+            if (!key.isEmpty() && key.startsWith("kafka.")) {
+                kafka09SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
+            }
+        }
         kafka09SinkTableInfo.check();
         return kafka09SinkTableInfo;
     }

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

@@ -44,10 +44,23 @@ public class KafkaSinkTableInfo extends TargetTableInfo {
 
 	private String topic;
 
+	public Map<String,String> kafkaParam = new HashMap<String,String>();
+
 	public KafkaSinkTableInfo() {
 		super.setType(CURR_TYPE);
 	}
 
+	public void addKafkaParam(String key,String value){
+		kafkaParam.put(key,value);
+	}
+
+	public String getKafkaParam(String key){
+		return kafkaParam.get(key);
+	}
+
+	public Set<String> getKafkaParamKeys(){
+		return kafkaParam.keySet();
+	}
 
 	public String getBootstrapServers() {
 		return bootstrapServers;

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

@@ -69,6 +69,9 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         this.fieldTypes = types;
 
         properties = new Properties();
+        for (String key:kafka10SinkTableInfo.getKafkaParamKeys()) {
+            properties.setProperty(key, kafka10SinkTableInfo.getKafkaParam(key));
+        }
         properties.setProperty("bootstrap.servers", kafka10SinkTableInfo.getBootstrapServers());
 
         this.serializationSchema = new JsonRowSerializationSchema(getOutputType());

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

@@ -42,6 +42,11 @@ 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())));
+        for (String key:props.keySet()) {
+            if (!key.isEmpty() && key.startsWith("kafka.")) {
+                kafka10SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
+            }
+        }
         kafka10SinkTableInfo.check();
         return kafka10SinkTableInfo;
     }

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

@@ -44,11 +44,24 @@ public class KafkaSinkTableInfo extends TargetTableInfo {
 
     private String bootstrapServers;
 
+    public Map<String,String> kafkaParam = new HashMap<String,String>();
+
     private String topic;
 
     public KafkaSinkTableInfo() {
         super.setType(CURR_TYPE);
     }
+    public void addKafkaParam(String key,String value){
+        kafkaParam.put(key,value);
+    }
+
+    public String getKafkaParam(String key){
+        return kafkaParam.get(key);
+    }
+
+    public Set<String> getKafkaParamKeys(){
+        return kafkaParam.keySet();
+    }
 
 
     public String getBootstrapServers() {

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

@@ -57,18 +57,20 @@ public class KafkaSink  implements AppendStreamTableSink<Row>, IStreamSinkGener<
 
     @Override
     public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
-        KafkaSinkTableInfo kafka10SinkTableInfo = (KafkaSinkTableInfo) targetTableInfo;
-        this.topic = kafka10SinkTableInfo.getTopic();
-        this.fieldNames = kafka10SinkTableInfo.getFields();
-        TypeInformation[] types = new TypeInformation[kafka10SinkTableInfo.getFields().length];
-        for (int i = 0; i < kafka10SinkTableInfo.getFieldClasses().length; i++) {
-            types[i] = TypeInformation.of(kafka10SinkTableInfo.getFieldClasses()[i]);
+        KafkaSinkTableInfo kafka11SinkTableInfo = (KafkaSinkTableInfo) targetTableInfo;
+        this.topic = kafka11SinkTableInfo.getTopic();
+        this.fieldNames = kafka11SinkTableInfo.getFields();
+        TypeInformation[] types = new TypeInformation[kafka11SinkTableInfo.getFields().length];
+        for (int i = 0; i < kafka11SinkTableInfo.getFieldClasses().length; i++) {
+            types[i] = TypeInformation.of(kafka11SinkTableInfo.getFieldClasses()[i]);
         }
         this.fieldTypes = types;
 
         properties = new Properties();
-        properties.setProperty("bootstrap.servers", kafka10SinkTableInfo.getBootstrapServers());
-
+        for (String key:kafka11SinkTableInfo.getKafkaParamKeys()) {
+            properties.setProperty(key, kafka11SinkTableInfo.getKafkaParam(key));
+        }
+        properties.setProperty("bootstrap.servers", kafka11SinkTableInfo.getBootstrapServers());
         this.serializationSchema = new JsonRowSerializationSchema(getOutputType());
         return this;
     }

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

@@ -42,6 +42,11 @@ 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())));
+        for (String key:props.keySet()) {
+            if (!key.isEmpty() && key.startsWith("kafka.")) {
+                kafka11SinkTableInfo.addKafkaParam(key.substring(6), props.get(key).toString());
+            }
+        }
         kafka11SinkTableInfo.check();
         return kafka11SinkTableInfo;
     }

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

@@ -44,10 +44,25 @@ public class KafkaSinkTableInfo extends TargetTableInfo {
 
     private String topic;
 
+    public Map<String,String> kafkaParam = new HashMap<String,String>();
+
     public KafkaSinkTableInfo() {
         super.setType(CURR_TYPE);
     }
 
+    public void addKafkaParam(String key,String value){
+        kafkaParam.put(key,value);
+    }
+
+    public String getKafkaParam(String key){
+        return kafkaParam.get(key);
+    }
+
+    public Set<String> getKafkaParamKeys(){
+        return kafkaParam.keySet();
+    }
+
+
 
     public String getBootstrapServers() {
         return bootstrapServers;