code: func (k *KafkaMdm) tryGetOffset(topic string, partition int32, offset int64, attempts int, sleep time.Duration) (int64, error) {
  
  	var val int64
  	var err error
  	var offsetStr string
  
  	switch offset {
  	case sarama.OffsetNewest:
  		offsetStr = "newest"
  	case sarama.OffsetOldest:
  		offsetStr = "oldest"
  	default:
  		offsetStr = strconv.FormatInt(offset, 10)
  	}
  
  	attempt := 1
  	for {
  		val, err = k.client.GetOffset(topic, partition, offset)
  		if err == nil {
  			break
  		}
  
  		err = fmt.Errorf("failed to get offset %s of partition %s:%d. %s (attempt %d/%d)", offsetStr, topic, partition, err, attempt, attempts)
  		if attempt == attempts {
  			break
  		}
  		log.Warnf("kafkamdm: %s", err.Error())
  		attempt += 1
  		time.Sleep(sleep)
  	}
  	return val, err
  }
docs: // tryGetOffset will to query kafka repeatedly for the requested offset and give up after attempts unsuccesfull attempts
  // an error is returned when it had to give up