protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc, Runnable runnable) {
        if (StringUtils.equalsIgnoreCase(offset, EKafkaOffset.EARLIEST.name())) {
            kafkaSrc.setStartFromEarliest();
        } else if(StringUtils.equalsIgnoreCase(offset, EKafkaOffset.TIMESTAMP.name())) {
            runnable.run();
        }else if (DtStringUtil.isJson(offset)) {
            Map<KafkaTopicPartition, Long> specificStartupOffsets = buildOffsetMap(offset, topicName);
            kafkaSrc.setStartFromSpecificOffsets(specificStartupOffsets);
        } else {
            kafkaSrc.setStartFromLatest();
        }
    }
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