Method starting

modules/generator/generator.go:117–146

(ctx context.Context)

}

func (g *Generator) starting(ctx context.Context) error {
	if g.cfg.ConsumeFromKafka {
		kafkaClient, err := ingest.NewGroupReaderClient(
			g.cfg.Ingest.Kafka,
			g.partitionRing,
			ingest.NewReaderClientMetrics("generator", prometheus.DefaultRegisterer),
			g.logger,
			kgo.InstanceID(g.cfg.InstanceID),
			kgo.OnPartitionsAssigned(func(_ context.Context, _ *kgo.Client, m map[string][]int32) {
				g.handlePartitionsAssigned(m)
			}),
			kgo.OnPartitionsRevoked(func(_ context.Context, _ *kgo.Client, m map[string][]int32) {
				g.handlePartitionsRevoked(m)
			}),
		)
		if err != nil {
			return fmt.Errorf("failed to create kafka reader client: %w", err)
		}

		g.kafkaClient = kafkaClient
		if err := ingest.WaitForKafkaBroker(ctx, g.kafkaClient.Client, g.logger); err != nil {
			return fmt.Errorf("failed to start metrics generator: %w", err)
		}

		g.kafkaAdm = kadm.NewClient(g.kafkaClient.Client)
		g.partitionClient = ingest.NewPartitionOffsetClient(g.kafkaClient.Client, g.cfg.Ingest.Kafka.Topic)
	}

	return nil
}

func (g *Generator) running(ctx context.Context) error {
	if g.cfg.ConsumeFromKafka {