// seekOffsets returns all valid offsets in partitions
seekOffsets returns all valid offsets in partitions
// seekOffsets returns all valid offsets in partitions
func (ks *KafkaSeeker) seekOffsets(topic string, partitions []int32, pos int64) ([]int64, error) {
	offsets := make([]int64, len(partitions))
	for _, partition := range partitions {
		start, err := ks.client.GetOffset(topic, partition, sarama.OffsetOldest)
		if err != nil {
			err = errors.Trace(err)
			return nil, err
		}
		end, err := ks.client.GetOffset(topic, partition, sarama.OffsetNewest)
		if err != nil {
			err = errors.Trace(err)
			return nil, err
		}
		offset, err := ks.seekOffset(topic, partition, start, end-1, pos)
		if err != nil {
			err = errors.Trace(err)
			return nil, err
		}
		offsets[partition] = offset
	}
	return offsets, nil
}
func ( ks * KafkaSeeker ) seekOffsets ( topic string , partitions [ ] int32 , pos int64 ) ( [ ] int64 , error ) { offsets := make ( [ ] int64 , len ( partitions ) ) for _ , partition := range partitions { start , err := ks . client . GetOffset ( topic , partition , sarama . OffsetOldest ) if err != nil { err = errors . Trace ( err ) return nil , err } end , err := ks . client . GetOffset ( topic , partition , sarama . OffsetNewest ) if err != nil { err = errors . Trace ( err ) return nil , err } offset , err := ks . seekOffset ( topic , partition , start , end - 1 , pos ) if err != nil { err = errors . Trace ( err ) return nil , err } offsets [ partition ] = offset } return offsets , nil }
seekOffsets returns all valid offsets in partitions
seekOffsets returns all valid offsets in partitions