Method starting

modules/blockbuilder/blockbuilder.go:183–223

(ctx context.Context)

}

func (b *BlockBuilder) starting(ctx context.Context) (err error) {
	level.Info(b.logger).Log("msg", "block builder starting")
	topic := b.cfg.IngestStorageConfig.Kafka.Topic

	b.enc, err = encoding.FromVersionForWrites(b.cfg.BlockConfig.Version)
	if err != nil {
		return fmt.Errorf("failed to create encoding: %w", err)
	}

	b.wal, err = wal.New(&b.cfg.WAL)
	if err != nil {
		return fmt.Errorf("failed to create WAL: %w", err)
	}

	b.kafkaClient, err = ingest.NewReaderClient(
		b.cfg.IngestStorageConfig.Kafka,
		ingest.NewReaderClientMetrics(blockBuilderServiceName, prometheus.DefaultRegisterer),
		b.logger,
	)
	if err != nil {
		return fmt.Errorf("failed to create kafka reader client: %w", err)
	}

	err = ingest.WaitForKafkaBroker(ctx, b.kafkaClient, b.logger)
	if err != nil {
		return fmt.Errorf("failed to start blockbuilder: %w", err)
	}

	b.partitionOffsetClient = ingest.NewPartitionOffsetClient(b.kafkaClient, topic)
	b.kadm = kadm.NewClient(b.kafkaClient)

	ingest.ExportPartitionLagMetrics(
		ctx,
		b.kafkaClient,
		b.logger,
		b.cfg.IngestStorageConfig,
		b.getAssignedPartitions,
		b.kafkaClient.ForceMetadataRefresh)

	return nil
}

func (b *BlockBuilder) running(ctx context.Context) error {
	defer close(b.consumeStopped)