Method readKafka

modules/generator/generator_kafka.go:98–131

(ctx context.Context)

}

func (g *Generator) readKafka(ctx context.Context) error {
	fetches := g.kafkaClient.PollFetches(ctx)
	fetches.EachError(func(_ string, _ int32, err error) {
		if !errors.Is(err, context.Canceled) {
			level.Error(g.logger).Log("msg", "failed to fetch records", "err", err)
		}
	})
	if err := fetches.Err(); err != nil && !errors.Is(err, context.Canceled) {
		return err
	}

	// Metric lag based on first message in each partition.
	// This balances overhead with granularity.
	fetches.EachPartition(func(p kgo.FetchTopicPartition) {
		if len(p.Records) > 0 {
			lag := time.Since(p.Records[0].Timestamp)
			ingest.SetPartitionLagSeconds(g.cfg.Ingest.Kafka.ConsumerGroup, p.Partition, lag)
		}
	})

	start := time.Now()

	for iter := fetches.RecordIter(); !iter.Done(); {
		select {
		case g.kafkaCh <- iter.Next():
		case <-ctx.Done():
			return ctx.Err()
		}
	}

	metricEnqueueTime.Add(time.Since(start).Seconds())

	return nil
}

// readCh reads records from the internal channel.
// This allows for offloading the expensive proto unmarshal