// Gets the offset for a given topic, partition and consumer group.
// Returns offset on sucess, error otherwise.
194140
func (this *ZookeeperCoordinator) GetOffset(Groupid string, topic string, partition int32) (offset int64, err error) {
	backoffMultiplier := 1
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		offset, err = this.tryGetOffsetForTopicPartition(Groupid, topic, partition)
		if err == nil {
			return
		}
		Tracef(this, "GetOffset for group %s, topic %s and partition %d failed after %d-th retry", Groupid, topic, partition, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
1031260
func (p *PdfDictionaryObj) GetOffset(glyph int) int {
	ttfp := p.PtrToSubsetFontObj.GetTTFParser()
	glyf := ttfp.GetTables()["glyf"]
	offset := int(glyf.Offset + ttfp.LocaTable[glyph])
	return offset
}
263479
func (bldr *aeBackupInformationKindFilesQueryBuilder) Offset(offset int) *aeBackupInformationKindFilesQueryBuilder {
	bldr.q = bldr.q.Offset(offset)
	if bldr.plugin != nil {
		bldr.plugin.Offset(offset)
	}
	return bldr
}
1132088
func (q *OffsetQueue) GetMaxOffset() uint64 {
	q.Lock()
	v := q.getMaxOffset0()
	q.Unlock()
	return v
}
1098191
func (c *KafkaClient) GetTopicMetadata(topics []string) (*MetadataResponse, error) {
	for i := 0; i <= c.config.MetadataRetries; i++ {
		if metadata, err := c.getMetadata(topics); err == nil {
			return metadata, nil
		}

		log.Debugf("GetTopicMetadata for %s failed after %d try", topics, i)
		time.Sleep(c.config.MetadataBackoff)
	}

	return nil, fmt.Errorf("Could not get topic metadata for %s after %d retries", topics, c.config.MetadataRetries)
}
1076157
func (idx *MemoryIndex) FindOffset(h plumbing.Hash) (int64, error) {
	if len(idx.FanoutMapping) <= int(h[0]) {
		return 0, plumbing.ErrObjectNotFound
	}

	k := idx.FanoutMapping[h[0]]
	i, ok := idx.findHashIndex(h)
	if !ok {
		return 0, plumbing.ErrObjectNotFound
	}

	offset := idx.getOffset(k, i)

	if !idx.offsetHashIsFull {
		// Save the offset for reverse lookup
		if idx.offsetHash == nil {
			idx.offsetHash = make(map[int64]plumbing.Hash)
		}
		idx.offsetHash[int64(offset)] = h
	}

	return int64(offset), nil
}
497573
func (t *Topic) Partition(id int32, replicas []int32) *Partition {
	return &Partition{ID: id, Replicas: replicas, topic: t}
}
968048
func (cg *Consumergroup) FetchOffset(topic string, partition int32) (int64, error) {
	node := fmt.Sprintf("%s/consumers/%s/offsets/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	val, _, err := cg.kz.conn.Get(node)
	if err == zk.ErrNoNode {
		return -1, nil
	} else if err != nil {
		return -1, err
	}
	return strconv.ParseInt(string(val), 10, 64)
}
694880
func (cg *Consumergroup) CommitOffset(topic string, partition int32, offset int64) error {
	node := fmt.Sprintf("%s/consumers/%s/offsets/%s/%d", cg.kz.conf.Chroot, cg.Name, topic, partition)
	data := []byte(fmt.Sprintf("%d", offset))

	_, stat, err := cg.kz.conn.Get(node)
	switch err {
	case zk.ErrNoNode: // Create a new node
		return cg.kz.create(node, data, false)

	case nil: // Update the existing node
		_, err := cg.kz.conn.Set(node, data, stat.Version)
		return err

	default:
		return err
	}
}
892272
func (q *OffsetQueue) Get(offset uint64) ([]interface{}, uint64) {
	oldOffset := offset
	if offset > 0 {
		offset = offset - 1
	}

	q.Lock()
	max := q.getMaxOffset0()
	if offset > max {
		panic(fmt.Sprintf("bug: error offset %d, end is %d", offset, q.end))
	} else if offset < q.start || (oldOffset == 0 && offset == q.start && q.start == 0) {
		value := q.items[0:]
		q.Unlock()
		return value, max
	}

	var value []interface{}
	for i := q.start; i < q.end; i++ {
		if i <= offset {
			q.items[i-q.start] = nil
		} else {
			value = append(value, q.items[i-q.start])
		}
	}

	old := q.start
	q.start = offset + 1
	if q.start < q.end {
		q.items = q.items[q.start-old:]
	} else {
		q.items = make([]interface{}, 0, 0)
	}
	q.Unlock()

	return value, max
}
1069721
func GetOffsetCtx(r *http.Request) int {
	offset, ok := r.Context().Value(ctxOffset).(int)
	if !ok {
		apiLog.Warn("offset is not set or is not an int")
		return 0
	}
	return offset
}
963560
func (s *storageImpl) getNextOffset(ctx context.Context, url string, length int64) (offset int64, err error) {
	r, err := http.NewRequest("PUT", url, nil)
	if err != nil {
		return
	}
	r.Header.Set("Content-Range", fmt.Sprintf("bytes */%d", length))
	r.Header.Set("Content-Length", "0")
	r.Header.Set("User-Agent", s.userAgent)
	resp, err := ctxhttp.Do(ctx, s.client, r)
	if err != nil {
		if isTemporaryNetError(err) {
			err = errTransientError
		}
		return
	}
	resp.Body.Close()

	if resp.StatusCode == 200 {
		// Fully uploaded.
		offset = length
	} else if resp.StatusCode == 308 {
		// Partially uploaded, extract last uploaded offset from Range header.
		rangeHeader := resp.Header.Get("Range")
		if rangeHeader != "" {
			_, err = fmt.Sscanf(rangeHeader, "bytes=0-%d", &offset)
			if err == nil {
				// |offset| is an *offset* of a last uploaded byte, not the data length.
				offset++
			}
		}
	} else if isTemporaryHTTPError(resp.StatusCode) {
		err = errTransientError
	} else {
		err = fmt.Errorf("unexpected response (HTTP %d) when querying for uploaded offset", resp.StatusCode)
	}
	return
}
1268112
func (bldr *aeBackupKindQueryBuilder) Offset(offset int) *aeBackupKindQueryBuilder {
	bldr.q = bldr.q.Offset(offset)
	if bldr.plugin != nil {
		bldr.plugin.Offset(offset)
	}
	return bldr
}
691420
func (c *Consumer) StartStaticPartitions(topicPartitionMap map[string][]int32) {
	topicsToNumStreamsMap := make(map[string]int)
	for topic := range topicPartitionMap {
		topicsToNumStreamsMap[topic] = c.config.NumConsumerFetchers
	}

	c.topicCount = &StaticTopicsToNumStreams{
		ConsumerId:            c.config.Consumerid,
		TopicsToNumStreamsMap: topicsToNumStreamsMap,
	}

	c.config.Coordinator.RegisterConsumer(c.config.Consumerid, c.config.Groupid, c.topicCount)
	allTopics, err := c.config.Coordinator.GetAllTopics()
	if err != nil {
		panic(err)
	}
	brokers, err := c.config.Coordinator.GetAllBrokers()
	if err != nil {
		panic(err)
	}

	time.Sleep(c.config.DeploymentTimeout)

	assignmentContext := newStaticAssignmentContext(c.config.Groupid, c.config.Consumerid, []string{c.config.Consumerid}, allTopics, brokers, c.topicCount, topicPartitionMap)
	partitionOwnershipDecision := newPartitionAssignor(c.config.PartitionAssignmentStrategy)(assignmentContext)

	topicPartitions := make([]*TopicAndPartition, 0)
	for topicPartition, _ := range partitionOwnershipDecision {
		topicPartitions = append(topicPartitions, &TopicAndPartition{topicPartition.Topic, topicPartition.Partition})
	}

	offsets, err := c.fetchOffsets(topicPartitions)
	if err != nil {
		panic(fmt.Sprintf("Failed to fetch offsets during rebalance: %s", err))
	}
	for _, topicPartition := range topicPartitions {
		offset := offsets[*topicPartition]
		threadId := partitionOwnershipDecision[*topicPartition]
		c.addPartitionTopicInfo(c.topicRegistry, topicPartition, offset, threadId)
	}

	if c.reflectPartitionOwnershipDecision(partitionOwnershipDecision) {
		c.updateFetcher(c.config.NumConsumerFetchers)
		c.initializeWorkerManagers()
	} else {
		panic("Could not reflect partition ownership")
	}

	go func() {
		Infof(c, "Restarted streams")
		c.connectChannels <- true
	}()

	c.startStreams()
}
563171
func (batch *Batch) Offset() int64 {
	batch.mutex.Lock()
	offset := batch.offset
	batch.mutex.Unlock()
	return offset
}
759772
func (bldr *SampleQueryBuilder) Offset(offset int) *SampleQueryBuilder {
	bldr.q = bldr.q.Offset(offset)
	if bldr.plugin != nil {
		bldr.plugin.Offset(offset)
	}
	return bldr
}
433467
func queryWatermarkOffsets(H Handle, topic string, partition int32, timeoutMs int) (low, high int64, err error) {
	h := H.gethandle()

	ctopic := C.CString(topic)
	defer C.free(unsafe.Pointer(ctopic))

	var cLow, cHigh C.int64_t

	e := C.rd_kafka_query_watermark_offsets(h.rk, ctopic, C.int32_t(partition),
		&cLow, &cHigh, C.int(timeoutMs))
	if e != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return 0, 0, newError(e)
	}

	low = int64(cLow)
	high = int64(cHigh)
	return low, high, nil
}