docs: private static void requestAndSetOffsetsFromKafka(
  			SimpleConsumer consumer,
  			List<KafkaTopicPartitionState<TopicAndPartition>> partitionStates,
  			Map<TopicAndPartition, PartitionOffsetRequestInfo> partitionToRequestInfo) throws IOException {
  		int retries = 0;
  		OffsetResponse response;
  		while (true) {
  			kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(
  				partitionToRequestInfo, kafka.api.OffsetRequest.CurrentVersion(), consumer.clientId());
  			response = consumer.getOffsetsBefore(request);
  
  			if (response.hasError()) {
  				StringBuilder exception = new StringBuilder();
  				for (KafkaTopicPartitionState<TopicAndPartition> part : partitionStates) {
  					short code;
  					if ((code = response.errorCode(part.getTopic(), part.getPartition())) != ErrorMapping.NoError()) {
  						exception.append("\nException for topic=").append(part.getTopic())
  							.append(" partition=").append(part.getPartition()).append(": ")
  							.append(ExceptionUtils.stringifyException(ErrorMapping.exceptionFor(code)));
  					}
  				}
  				if (++retries >= 3) {
  					throw new IOException("Unable to get last offset for partitions " + partitionStates + ": "
  						+ exception.toString());
  				} else {
  					LOG.warn("Unable to get last offset for partitions: Exception(s): {}", exception);
  				}
  			} else {
  				break; // leave retry loop
  			}
  		}
  
  		for (KafkaTopicPartitionState<TopicAndPartition> part: partitionStates) {
  			// there will be offsets only for partitions that were requested for
  			if (partitionToRequestInfo.containsKey(part.getKafkaPartitionHandle())) {
  				final long offset = response.offsets(part.getTopic(), part.getPartition())[0];
  
  				// the offset returned is that of the next record to fetch. because our state reflects the latest
  				// successfully emitted record, we subtract one
  				part.setOffset(offset - 1);
  			}
  		}
  	}