kafka09 sink

Author: todd5167

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

@@ -20,12 +20,12 @@
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.connectors.kafka.*;
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase;
+import org.apache.flink.streaming.connectors.kafka.Kafka09TableSink;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
-import org.apache.flink.table.util.TableConnectorUtil;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.utils.TableConnectorUtils;
 import org.apache.flink.types.Row;
 
 import java.util.Optional;
@@ -38,30 +38,26 @@
  *
  * @author maqi
  */
-public class CustomerKafka09JsonTableSink extends KafkaJsonTableSink {
+public class CustomerKafka09JsonTableSink extends Kafka09TableSink {
 
 
 	protected SerializationSchema schema;
 
-	public CustomerKafka09JsonTableSink(String topic, Properties properties, SerializationSchema schema) {
-		super(topic, properties, new FlinkFixedPartitioner<>());
-		this.schema = schema;
-	}
 
-	public CustomerKafka09JsonTableSink(String topic, Properties properties, FlinkKafkaPartitioner<Row> partitioner, SerializationSchema schema) {
-		super(topic, properties, partitioner);
-		this.schema = schema;
-	}
 
 
 	@Deprecated
-	public CustomerKafka09JsonTableSink(String topic, Properties properties, KafkaPartitioner<Row> partitioner, SerializationSchema schema) {
-		super(topic, properties, new FlinkKafkaDelegatePartitioner<>(partitioner));
-		this.schema = schema;
+	public CustomerKafka09JsonTableSink(TableSchema schema,
+										String topic,
+										Properties properties,
+										Optional<FlinkKafkaPartitioner<Row>> partitioner,
+										SerializationSchema<Row> serializationSchema) {
+		super(schema, topic, properties, partitioner, serializationSchema);
+		this.schema = serializationSchema;
 	}
 
 	@Override
-	protected SinkFunction<Row> createKafkaProducer(String s, Properties properties, SerializationSchema<Row> serializationSchema, Optional<FlinkKafkaPartitioner<Row>> optional) {
+	protected FlinkKafkaProducerBase<Row> createKafkaProducer(String s, Properties properties, SerializationSchema<Row> serializationSchema, Optional<FlinkKafkaPartitioner<Row>> optional) {
 		return new CustomerFlinkKafkaProducer09<>(topic, serializationSchema, properties);
 	}
 
@@ -70,6 +66,6 @@ public void emitDataStream(DataStream<Row> dataStream) {
 		SinkFunction<Row> kafkaProducer = createKafkaProducer(topic, properties, schema, partitioner);
 		// always enable flush on checkpoint to achieve at-least-once if query runs with checkpointing enabled.
 		//kafkaProducer.setFlushOnCheckpoint(true);
-		dataStream.addSink(kafkaProducer).name(TableConnectorUtil.generateRuntimeName(this.getClass(), fieldNames));
+		dataStream.addSink(kafkaProducer).name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
 	}
 }

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

@@ -26,9 +26,14 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.connectors.kafka.KafkaTableSinkBase;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
+import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
+import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.sinks.TableSink;
 import org.apache.flink.types.Row;
+
+import java.util.Optional;
 import java.util.Properties;
 
 /**
@@ -44,39 +49,63 @@ public class KafkaSink implements AppendStreamTableSink<Row>, IStreamSinkGener<K
 
 	protected TypeInformation<?>[] fieldTypes;
 
+	/** The schema of the table. */
+	private TableSchema schema;
+
+	/** The Kafka topic to write to. */
 	protected String topic;
 
+	/** Properties for the Kafka producer. */
 	protected Properties properties;
 
 	/** Serialization schema for encoding records to Kafka. */
 	protected SerializationSchema serializationSchema;
 
+	/** Partitioner to select Kafka partition for each item. */
+	protected Optional<FlinkKafkaPartitioner<Row>> partitioner;
+
 	@Override
 	public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
+
 		KafkaSinkTableInfo kafka09SinkTableInfo = (KafkaSinkTableInfo) targetTableInfo;
 		this.topic = kafka09SinkTableInfo.getTopic();
+
+		Properties props = new Properties();
+		props.setProperty("bootstrap.servers", kafka09SinkTableInfo.getBootstrapServers());
+
+		for (String key:kafka09SinkTableInfo.getKafkaParamKeys()) {
+			props.setProperty(key, kafka09SinkTableInfo.getKafkaParam(key));
+		}
+		this.properties = props;
+		this.partitioner = Optional.of(new FlinkFixedPartitioner<>());
 		this.fieldNames = kafka09SinkTableInfo.getFields();
 		TypeInformation[] types = new TypeInformation[kafka09SinkTableInfo.getFields().length];
-		for (int i = 0; i < kafka09SinkTableInfo.getFieldClasses().length; i++) {
+		for(int i = 0; i< kafka09SinkTableInfo.getFieldClasses().length; i++){
 			types[i] = TypeInformation.of(kafka09SinkTableInfo.getFieldClasses()[i]);
 		}
 		this.fieldTypes = types;
 
-		properties = new Properties();
-		for (String key : kafka09SinkTableInfo.getKafkaParamKeys()) {
-			properties.setProperty(key, kafka09SinkTableInfo.getKafkaParam(key));
+		TableSchema.Builder schemaBuilder = TableSchema.builder();
+		for (int i=0;i<fieldNames.length;i++){
+			schemaBuilder.field(fieldNames[i], fieldTypes[i]);
+		}
+		this.schema = schemaBuilder.build();
+
+		//this.serializationSchema = Optional.of(JsonRowSerializationSchema.class);
+		if ("json".equalsIgnoreCase(kafka09SinkTableInfo.getSinkDataType())) {
+			this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType());
 		}
-		properties.setProperty("bootstrap.servers", kafka09SinkTableInfo.getBootstrapServers());
 
-		this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType());
 		return this;
 	}
 
 	@Override
 	public void emitDataStream(DataStream<Row> dataStream) {
 		KafkaTableSinkBase kafkaTableSink = new CustomerKafka09JsonTableSink(
+				schema,
 				topic,
 				properties,
+				partitioner,
 				serializationSchema
 		);
 

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

@@ -36,6 +36,10 @@ public class KafkaSinkTableInfo extends TargetTableInfo {
 	//version
 	private static final String CURR_TYPE = "kafka09";
 
+	public KafkaSinkTableInfo() {
+		super.setType(CURR_TYPE);
+	}
+
 	public static final String BOOTSTRAPSERVERS_KEY = "bootstrapServers";
 
 	public static final String TOPIC_KEY = "topic";
@@ -46,9 +50,6 @@ public class KafkaSinkTableInfo extends TargetTableInfo {
 
 	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);

kafka09/pom.xml

@@ -15,7 +15,7 @@
 
     <modules>
         <module>kafka09-source</module>
-        <!--<module>kafka09-sink</module>-->
+        <module>kafka09-sink</module>
     </modules>
 
     <dependencies>