Method fetchPartitions

modules/blockbuilder/blockbuilder.go:546–576

It fetches all the offsets for the blockbuilder topic, for each owned partitions it calculates their last committed records and the end record offset. Based on that it sort the partitions by lag

(ctx context.Context, partitions []int32)

// It fetches all the offsets for the blockbuilder topic, for each owned partitions it calculates their last committed records and the
// end record offset. Based on that it sort the partitions by lag
func (b *BlockBuilder) fetchPartitions(ctx context.Context, partitions []int32) ([]partitionState, error) {
	var (
		ps          = make([]partitionState, 0, len(partitions))
		commits     kadm.OffsetResponses
		endsOffsets kadm.ListedOffsets
		err         error
	)

	boff := backoff.New(ctx, backoff.Config{
		MinBackoff: 100 * time.Millisecond,
		MaxBackoff: time.Minute,
		MaxRetries: 5,
	})
	for boff.Ongoing() {
		commits, endsOffsets, err = b.getPartitionOffsets(ctx, partitions)
		if err == nil {
			break
		}
		ingest.HandleKafkaError(err, b.kafkaClient.ForceMetadataRefresh)
		boff.Wait()
	}
	if err != nil {
		return nil, fmt.Errorf("failed to fetch partition offsets: %w", err)
	}
	for _, partition := range partitions {
		p := b.getPartitionState(partition, commits, endsOffsets)
		ps = append(ps, p)
	}

	return ps, nil
}

// todo: this function fetches the offsets for all the partitions including the ones that are not assigned to this block builder.
// improve it to only fetch the offsets for the assigned partitions