Method readCh

modules/generator/generator_kafka.go:136–178

readCh reads records from the internal channel. This allows for offloading the expensive proto unmarshal to multiple goroutines.

(ctx context.Context)

// This allows for offloading the expensive proto unmarshal
// to multiple goroutines.
func (g *Generator) readCh(ctx context.Context) {
	defer g.kafkaWG.Done()

	var c ingest.GeneratorCodec
	switch g.cfg.Codec {
	case codecPushBytes:
		c = ingest.NewPushBytesDecoder()
	case codecOTLP:
		c = ingest.NewOTLPDecoder()
	}

	for {
		var r *kgo.Record
		select {
		case r = <-g.kafkaCh:
		case <-ctx.Done():
			return
		}

		tenant := string(r.Key)

		i, err := g.getOrCreateInstance(tenant)
		if err != nil {
			level.Error(g.logger).Log("tenant", tenant, "msg", "consumeKafkaChannel getOrCreateInstance", "err", err)
			continue
		}

		iterator, err := c.Decode(r.Value)
		if err != nil {
			level.Error(g.logger).Log("tenant", tenant, "msg", "consumeKafkaChannel decode", "err", err)
			continue
		}

		for resourceSpans, err := range iterator {
			if err != nil {
				level.Error(g.logger).Log("tenant", tenant, "msg", "consumeKafkaChannel unmarshal", "err", err)
				continue
			}

			i.pushSpansFromQueue(ctx, r.Timestamp, resourceSpans)
		}
	}
}

func (g *Generator) getAssignedActivePartitions() []int32 {
	g.partitionMtx.Lock()