protected Properties getKafkaProperties(KafkaSourceTableInfo kafkaSourceTableInfo) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaSourceTableInfo.getBootstrapServers());

        if (DtStringUtil.isJson(kafkaSourceTableInfo.getOffsetReset())) {
            props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.NONE.name().toLowerCase());
        } else if(StringUtils.equalsIgnoreCase(EKafkaOffset.TIMESTAMP.name().toLowerCase(),  kafkaSourceTableInfo.getOffsetReset())){
                props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.EARLIEST.name().toLowerCase());
        } else {
            props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSourceTableInfo.getOffsetReset());
        }

        if (StringUtils.isNotBlank(kafkaSourceTableInfo.getGroupId())) {
            props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, kafkaSourceTableInfo.getGroupId());
        }
        if("kafka".equalsIgnoreCase(kafkaSourceTableInfo.getType())){
            String keyDeserializer = kafkaSourceTableInfo.getKafkaParam(KafkaSourceTableInfo.KEY_DESERIALIZER);
            if(StringUtils.isNotBlank(keyDeserializer)){
                kafkaSourceTableInfo.putKafkaParam(KafkaSourceTableInfo.KEY_DESERIALIZER, KafkaSourceTableInfo.DT_DESERIALIZER_CLASS_NAME);
                kafkaSourceTableInfo.putKafkaParam(KafkaSourceTableInfo.DT_KEY_DESERIALIZER, keyDeserializer);
            }
            String valueDeserializer = kafkaSourceTableInfo.getKafkaParam(KafkaSourceTableInfo.VALUE_DESERIALIZER);
            if(StringUtils.isNotBlank(valueDeserializer)){
                kafkaSourceTableInfo.putKafkaParam(KafkaSourceTableInfo.VALUE_DESERIALIZER, KafkaSourceTableInfo.DT_DESERIALIZER_CLASS_NAME);
                kafkaSourceTableInfo.putKafkaParam(KafkaSourceTableInfo.DT_VALUE_DESERIALIZER, valueDeserializer);
            }
        }
        for (String key : kafkaSourceTableInfo.getKafkaParamKeys()) {
            props.setProperty(key, kafkaSourceTableInfo.getKafkaParam(key));
        }
        return props;
    }
Date: 2020/3/20
Company: www.dtstack.com
@author maqi
kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java