Method consume

modules/blockbuilder/blockbuilder.go:262–331

It consumes records for all the asigneed partitions, priorizing the ones with more lag. It keeps consuming until all the partitions lag is less than the cycle duration. When that happen it returns time to wait before another consuming cycle, based on the last record timestamp

(ctx context.Context)

// It consumes records for all the asigneed partitions, priorizing the ones with more lag. It keeps consuming until
// all the partitions lag is less than the cycle duration. When that happen it returns time to wait before another consuming cycle, based on the last record timestamp
func (b *BlockBuilder) consume(ctx context.Context) (time.Duration, error) {
	partitions := b.getAssignedPartitions()

	ctx, span := tracer.Start(ctx, "blockbuilder.consume", trace.WithAttributes(attribute.String("active_partitions", formatActivePartitions(partitions))))
	defer span.End()

	if len(partitions) == 0 {
		return b.cfg.ConsumeCycleDuration, errNoPartitionsAssigned
	}

	level.Info(b.logger).Log("msg", "starting consume cycle", "active_partitions", formatActivePartitions(partitions))
	defer func(t time.Time) { metricConsumeCycleDuration.Observe(time.Since(t).Seconds()) }(time.Now())

	// Clear all previous remnants
	err := b.wal.Clear()
	if err != nil {
		return 0, err
	}

	ps, err := b.fetchPartitions(ctx, partitions)
	if err != nil {
		return 0, err
	}

	// First iteration over all the assigned partitions to get their current lag in time
	for i, p := range ps {
		if !p.hasRecords() { // No records, skip for the first iteration
			// We treat the partition as updated through now,
			// and will check it again after ConsumeCycleDuration has elapsed
			ps[i].lastRecordTs = time.Now()
			ps[i].commitOffset = 0 // always start at beginning
			level.Info(b.logger).Log("msg", "partition has no records", "partition", p.partition)
			continue
		}
		lastRecordTs, commitOffset, err := b.consumePartition(ctx, p)
		if err != nil {
			return 0, err
		}
		ps[i].lastRecordTs = lastRecordTs
		ps[i].commitOffset = commitOffset
	}

	// Iterate over the laggiest partition until the lag is less than the cycle duration or none of the partitions has records
	for {
		sort.Slice(ps, func(i, j int) bool {
			return ps[i].lastRecordTs.Before(ps[j].lastRecordTs)
		})

		laggiestPartition := ps[0]
		if laggiestPartition.lastRecordTs.IsZero() {
			return b.cfg.ConsumeCycleDuration, errors.New("partition has no last record timestamp")
		}

		lagTime := time.Since(laggiestPartition.lastRecordTs)
		if lagTime < b.cfg.ConsumeCycleDuration {
			return b.cfg.ConsumeCycleDuration - lagTime, nil
		}
		// If we don't know exact offset, we need to start over on next cycle to fetch offsets again.