@Override
    public FlinkKafkaConsumerBase<Row> createKafkaTableSource(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation, Properties props) {
        KafkaConsumer09 kafkaSrc;
        if (kafkaSourceTableInfo.getTopicIsPattern()) {
            DeserializationMetricWrapper deserMetricWrapper = createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> 0L);
            kafkaSrc =
                    new KafkaConsumer09(
                            Pattern.compile(kafkaSourceTableInfo.getTopic()),
                            kafkaSourceTableInfo.getSampleSize(),
                            deserMetricWrapper,
                            props);
        } else {
            DeserializationMetricWrapper deserMetricWrapper = createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> 0L);
            kafkaSrc =
                    new KafkaConsumer09(
                            kafkaSourceTableInfo.getTopic(),
                            kafkaSourceTableInfo.getSampleSize(),
                            deserMetricWrapper,
                            props);
        }
        return kafkaSrc;
    }
company: www.dtstack.com

@author: toutian
create: 2019/12/24
kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer09Factory.java