code: private void resetCorrectOffsets() {
  		String kafkaServers = consumerContext.getProperties().getProperty("bootstrap.servers");
  		String zkServers = consumerContext.getProperties().getProperty("zookeeper.connect");
  		if(StringUtils.isAnyBlank(kafkaServers,zkServers)){
  			logger.warn("resetCorrectOffsets exit。Please check [bootstrap.servers] and [zookeeper.connect] is existing");
  			return;
  		}
  		ZkConsumerCommand command = new ZkConsumerCommand(zkServers, kafkaServers);
  		try {
  			List<String> topics = command.getSubscribeTopics(consumerContext.getGroupId());
  			for (String topic : topics) {
  				List<TopicPartitionInfo> partitions = command.getTopicOffsets(consumerContext.getGroupId(), topic);
  				for (TopicPartitionInfo partition : partitions) {
  					//期望的偏移
  					long expectOffsets = consumerContext.getLatestProcessedOffsets(topic, partition.getPartition());
  					//
  					if(expectOffsets > 0 && expectOffsets < partition.getOffset()){			
  						command.resetTopicOffsets(consumerContext.getGroupId(), topic, partition.getPartition(), expectOffsets);
  						logger.info("seek Topic[{}] partition[{}] from {} to {}",topic,partition.getPartition(),partition.getOffset(),expectOffsets);
  					}
  				}
  			}
  		} catch (Exception e) {
  			e.printStackTrace();
  		}
  		command.close();
  	}
docs: 按上次记录重置offsets