protected Map<KafkaTopicPartition, Long> buildOffsetMap(String offsetJson, String topicName) {
        try {
            Properties properties = PluginUtil.jsonStrToObject(offsetJson, Properties.class);
            Map<String, Object> offsetMap = PluginUtil.objectToMap(properties);
            Map<KafkaTopicPartition, Long> specificStartupOffsets = offsetMap
                    .entrySet()
                    .stream()
                    .collect(Collectors.toMap(
                            (Map.Entry<String, Object> entry) -> new KafkaTopicPartition(topicName, Integer.valueOf(entry.getKey())),
                            (Map.Entry<String, Object> entry) -> Long.valueOf(entry.getValue().toString()))
                    );

            return specificStartupOffsets;
        } catch (Exception e) {
            throw new RuntimeException("not support offsetReset type:" + offsetJson);
        }
    }
kafka offset,eg.. {"0":12312,"1":12321,"2":12312}
@param offsetJson
@param topicName
@return
kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java