@Override
    public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        KafkaSourceTableInfo kafkaSourceTableInfo = (KafkaSourceTableInfo) sourceTableInfo;
        String topicName = kafkaSourceTableInfo.getTopic();

        Properties kafkaProperties = getKafkaProperties(kafkaSourceTableInfo);
        TypeInformation<Row> typeInformation = getRowTypeInformation(kafkaSourceTableInfo);
        FlinkKafkaConsumer09<Row> kafkaSrc = (FlinkKafkaConsumer09<Row>) new KafkaConsumer09Factory().createKafkaTableSource(kafkaSourceTableInfo, typeInformation, kafkaProperties);

        String sourceOperatorName = generateOperatorName(sourceTableInfo.getName(), topicName);
        DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);

        setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromLatest());
        String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");

        return tableEnv.fromDataStream(kafkaSource, fields);
    }
If eventtime field is specified, the default time field rowtime
Date: 2017/2/20
Company: www.dtstack.com
@author xuchao
kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java