@Override
    public KafkaSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
        KafkaSinkTableInfo kafka10SinkTableInfo = (KafkaSinkTableInfo) targetTableInfo;

        Properties kafkaProperties = getKafkaProperties(kafka10SinkTableInfo);
        this.tableName = kafka10SinkTableInfo.getName();
        this.updateMode = kafka10SinkTableInfo.getUpdateMode();
        this.topic = kafka10SinkTableInfo.getTopic();
        this.partitioner = Optional.of(new CustomerFlinkPartition<>());
        this.partitionKeys = getPartitionKeys(kafka10SinkTableInfo);
        this.fieldNames = kafka10SinkTableInfo.getFields();
        this.fieldTypes = getTypeInformations(kafka10SinkTableInfo);
        this.schema = buildTableSchema(fieldNames, fieldTypes);
        this.parallelism = kafka10SinkTableInfo.getParallelism();
        this.sinkOperatorName = SINK_OPERATOR_NAME_TPL.replace("${topic}", topic).replace("${table}", tableName);
        this.kafkaProducer011 = new KafkaProducer010Factory().createKafkaProducer(kafka10SinkTableInfo, getOutputType(), kafkaProperties, partitioner, partitionKeys);
        return this;
    }
Date: 2018/12/18
Company: www.dtstack.com

@author DocLi

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