// GetOffset gets the offset for a given group, topic and partition from Kafka. A part of new offset management API.
767674
func (c *KafkaClient) GetOffset(group string, topic string, partition int32) (int64, error) {
	log.Infof("Getting offset for group %s, topic %s, partition %d", group, topic, partition)
	coordinator, err := c.metadata.OffsetCoordinator(group)
	if err != nil {
		return InvalidOffset, err
	}

	request := NewOffsetFetchRequest(group)
	request.AddOffset(topic, partition)
	bytes, err := c.syncSendAndReceive(coordinator, request)
	if err != nil {
		return InvalidOffset, err
	}
	response := new(OffsetFetchResponse)
	decodingErr := c.decode(bytes, response)
	if decodingErr != nil {
		log.Errorf("Could not decode an OffsetFetchResponse. Reason: %s", decodingErr.Reason())
		return InvalidOffset, decodingErr.Error()
	}

	topicOffsets, exist := response.Offsets[topic]
	if !exist {
		return InvalidOffset, fmt.Errorf("OffsetFetchResponse does not contain information about requested topic")
	}

	if offset, exists := topicOffsets[partition]; !exists {
		return InvalidOffset, fmt.Errorf("OffsetFetchResponse does not contain information about requested partition")
	} else if offset.Error != ErrNoError {
		return InvalidOffset, offset.Error
	} else {
		return offset.Offset, nil
	}
}
1272908
func (c *Consumer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error) {
	return queryWatermarkOffsets(c, topic, partition, timeoutMs)
}
581052
func (d *Decoder) savedOffset() int {
	n := d.saved.Len()
	if d.nextByte >= 0 {
		n--
	}
	return n
}
378500
func (a *DateHistogramAggregation) Offset(offset string) *DateHistogramAggregation {
	a.offset = offset
	return a
}
642463
func (m *LogManager) CommitOffset(o Offset, override bool) error {
	m.Lock()
	defer m.Unlock()
	if currentOffset, ok := m.nsMap[o.Namespace]; !override && ok && currentOffset >= o.LogOffset {
		log.With("currentOffest", currentOffset).
			With("providedOffset", o.LogOffset).
			Debugln("refusing to commit offset")
		return nil
	}
	_, err := m.log.Append(o.Bytes())
	if err != nil {
		return err
	}
	m.nsMap[o.Namespace] = o.LogOffset
	return nil
}
971619
func (b *Broker) offset(topic string, partition int32, timems int64) (offset int64, err error) {
	for retry := 0; retry < b.conf.RetryErrLimit; retry++ {
		if retry != 0 {
			time.Sleep(b.conf.RetryErrWait)
			err = b.muRefreshMetadata()
			if err != nil {
				continue
			}
		}
		var conn *connection
		conn, err = b.muLeaderConnection(topic, partition)
		if err != nil {
			return 0, err
		}
		var resp *proto.OffsetResp
		resp, err = conn.Offset(&proto.OffsetReq{
			RequestHeader: proto.RequestHeader{ClientID: b.conf.ClientID},
			ReplicaID:     -1, // any client
			Topics: []proto.OffsetReqTopic{
				{
					Name: topic,
					Partitions: []proto.OffsetReqPartition{
						{
							ID:         partition,
							TimeMs:     timems,
							MaxOffsets: 2,
						},
					},
				},
			},
		})
		if err != nil {
			if _, ok := err.(*net.OpError); ok || err == io.EOF || err == syscall.EPIPE {
				// Connection is broken, so should be closed, but the error is
				// still valid and should be returned so that retry mechanism have
				// chance to react.
				b.conf.Logger.Debug("connection died while sending message",
					"topic", topic,
					"partition", partition,
					"error", err)
				b.muCloseDeadConnection(conn)
			}
			continue
		}
		for _, t := range resp.Topics {
			if t.Name != topic {
				b.conf.Logger.Debug("unexpected topic information",
					"expected", topic,
					"got", t.Name)
				continue
			}
			for _, part := range t.Partitions {
				if part.ID != partition {
					b.conf.Logger.Debug("unexpected partition information",
						"topic", t.Name,
						"expected", partition,
						"got", part.ID)
					continue
				}
				if err = part.Err; err == nil {
					if len(part.Offsets) == 0 {
						return 0, nil
					} else {
						return part.Offsets[0], nil
					}
				}
			}
		}
	}
	return 0, errors.New("incomplete fetch response")
}
590155
func (br *bitReader) FlushOffset() int64 {
	if br.bufRd == nil {
		return br.offset
	}

	// Update the number of total bits to discard.
	br.discardBits += int(br.fedBits - br.numBits)
	br.fedBits = br.numBits

	// Discard some bytes to update read offset.
	nd := (br.discardBits + 7) / 8 // Round up to nearest byte
	nd, _ = br.bufRd.Discard(nd)
	br.discardBits -= nd * 8 // -7..0
	br.offset += int64(nd)

	// These are invalid after Discard.
	br.bufPeek = nil
	return br.offset
}
968156
func (f *File) DataOffset() (offset int64, err error) {
	bodyOffset, err := f.findBodyOffset()
	if err != nil {
		return
	}
	return f.headerOffset + bodyOffset, nil
}
1119361
func (s *Stack) PeekByOffset(offset uint64) (*Item, error) {
	s.RLock()
	defer s.RUnlock()

	// Check if stack is closed.
	if !s.isOpen {
		return nil, ErrDBClosed
	}

	return s.getItemByID(s.head - offset)
}
417234
func (s *ScanOp) Offset(token string) *ScanOp {
	s.token = token
	return s
}
407327
func (this *SiestaClient) CommitOffset(group string, topic string, partition int32, offset int64) error {
	return this.connector.CommitOffset(group, topic, partition, offset)
}
176616
func (o *ListOAuth2ClientsParams) WithOffset(offset *int64) *ListOAuth2ClientsParams {
	o.SetOffset(offset)
	return o
}
310987
func (b *Broker) FetchOffset(request *OffsetFetchRequest) (*OffsetFetchResponse, error) {
	response := new(OffsetFetchResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}
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
}
338641
func ExtractOffset(opts ...interface{}) *opt.OffsetOption {
	for _, o := range opts {
		if v, ok := o.(*opt.OffsetOption); ok {
			return v
		}
	}
	return nil
}
798994
func (p *Packfile) GetSizeByOffset(o int64) (size int64, err error) {
	if _, err := p.s.SeekFromStart(o); err != nil {
		if err == io.EOF || isInvalid(err) {
			return 0, plumbing.ErrObjectNotFound
		}

		return 0, err
	}

	h, err := p.nextObjectHeader()
	if err != nil {
		return 0, err
	}
	return p.getObjectSize(h)
}
692204
func (p *contentProvider) findOffset(filename bool, r uint32) uint32 {
	if p.id.metaData.PlainASCII {
		return r
	}

	sample := p.id.runeOffsets
	runeEnds := p.id.fileEndRunes
	fileStartByte := p.id.boundaries[p.idx]
	if filename {
		sample = p.id.fileNameRuneOffsets
		runeEnds = p.id.fileNameEndRunes
		fileStartByte = p.id.fileNameIndex[p.idx]
	}

	absR := r
	if p.idx > 0 {
		absR += runeEnds[p.idx-1]
	}

	byteOff := sample[absR/runeOffsetFrequency]
	left := absR % runeOffsetFrequency

	var data []byte

	if filename {
		data = p.id.fileNameContent[byteOff:]
	} else {
		data, p.err = p.id.readContentSlice(byteOff, 3*runeOffsetFrequency)
		if p.err != nil {
			return 0
		}
	}
	for left > 0 {
		_, sz := utf8.DecodeRune(data)
		byteOff += uint32(sz)
		data = data[sz:]
		left--
	}

	byteOff -= fileStartByte
	return byteOff
}