code: private void resetCorrectOffsets(ConsumerWorker worker) {	
  		
  		KafkaConsumer<String, Serializable> consumer = worker.consumer;
  		Map<String, List<PartitionInfo>> topicInfos = consumer.listTopics();
  		Set<String> topics = topicInfos.keySet();
  		
  		List<String> expectTopics = new ArrayList<>(topicHandlers.keySet());
  		
  		List<PartitionInfo> patitions = null;
  		
  		consumer.poll(200);
  		
  		for (String topic : topics) {
  			if(!expectTopics.contains(topic))continue;
  			
  			patitions = topicInfos.get(topic);
  			for (PartitionInfo partition : patitions) {
  				try {						
  					//期望的偏移
  					long expectOffsets = consumerContext.getLatestProcessedOffsets(topic, partition.partition());
  					//
  					TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
  					OffsetAndMetadata metadata = consumer.committed(topicPartition);
  					
  					Set<TopicPartition> assignment = consumer.assignment();
  					if(assignment.contains(topicPartition)){
  						if(expectOffsets > 0 && expectOffsets < metadata.offset()){								
  							consumer.seek(topicPartition, expectOffsets);
  							//consumer.seekToBeginning(assignment);
  					        logger.info(">>>>>>> seek Topic[{}] partition[{}] from {} to {}",topic, partition.partition(),metadata.offset(),expectOffsets);
  						}
  					}
  				} catch (Exception e) {
  					logger.warn("try seek topic["+topic+"] partition["+partition.partition()+"] offsets error");
  				}
  			}
  		}
  		consumer.resume(consumer.assignment());
  	}
docs: 按上次记录重置offsets