Method consumePartition

modules/blockbuilder/blockbuilder.go:333–503

(ctx context.Context, ps partitionState)

}

func (b *BlockBuilder) consumePartition(ctx context.Context, ps partitionState) (lastTs time.Time, commitOffset int64, err error) {
	ctx, span := tracer.Start(ctx, "blockbuilder.consumePartition",
		trace.WithAttributes(attribute.Int("partition", int(ps.partition)),
			attribute.String("last_record_ts", ps.lastRecordTs.String())))

	defer func(t time.Time) {
		metricProcessPartitionSectionDuration.WithLabelValues(strconv.Itoa(int(ps.partition))).Observe(time.Since(t).Seconds())
		span.End()
	}(time.Now())

	var (
		dur              = b.cfg.ConsumeCycleDuration
		topic            = b.cfg.IngestStorageConfig.Kafka.Topic
		group            = b.cfg.IngestStorageConfig.Kafka.ConsumerGroup
		maxBytesPerCycle = b.cfg.MaxBytesPerCycle
		partLabel        = strconv.Itoa(int(ps.partition))
		consumedBytes    uint64
		startOffset      kgo.Offset
		init             bool
		writer           *writer
		lastRec          *kgo.Record
		end              time.Time
		processedRecords int
	)

	startOffset = ps.getStartOffset()

	level.Info(b.logger).Log(
		"msg", "consuming partition",
		"partition", ps.partition,
		"commit_offset", ps.commitOffset,
		"start_offset", startOffset,
	)
	// We always rewind the partition's offset to the commit offset by reassigning the partition to the client (this triggers partition assignment).
	// This is so the cycle started exactly at the commit offset, and not at what was (potentially over-) consumed previously.
	// In the end, we remove the partition from the client (refer to the defer below) to guarantee the client always consumes
	// from one partition at a time. I.e. when this partition is consumed, we start consuming the next one.
	b.kafkaClient.AddConsumePartitions(map[string]map[int32]kgo.Offset{
		topic: {
			ps.partition: startOffset,
		},
	})
	defer b.kafkaClient.RemoveConsumePartitions(map[string][]int32{topic: {ps.partition}})

outer:
	for {
		fetches := func() kgo.Fetches {
			defer func(t time.Time) { metricFetchDuration.WithLabelValues(partLabel).Observe(time.Since(t).Seconds()) }(time.Now())
			ctx2, cancel := context.WithTimeout(ctx, pollTimeout)
			defer cancel()
			return b.kafkaClient.PollFetches(ctx2)
		}()
		err = fetches.Err()
		if err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				// No more data
				break
			}