Method consume

modules/livestore/live_store.go:625–688

(ctx context.Context, rs recordIter, now time.Time)

}

func (s *LiveStore) consume(ctx context.Context, rs recordIter, now time.Time) (*kadm.Offset, error) {
	defer s.decoder.Reset()
	ctx, span := tracer.Start(ctx, "LiveStore.consume")
	defer span.End()

	recordCount := 0
	var lastRecord *kgo.Record

	cutoff := now.Add(-s.cfg.CompleteBlockTimeout)
	// Process records by tenant
	for !rs.Done() {
		record := rs.Next()
		tenant := string(record.Key)

		// Track partition lag in seconds
		lag := now.Sub(record.Timestamp)
		ingest.SetPartitionLagSeconds(s.cfg.IngestConfig.Kafka.ConsumerGroup, record.Partition, lag)

		if record.Timestamp.Before(cutoff) {
			metricRecordsDropped.WithLabelValues(tenant, droppedRecordReasonTooOld).Inc()
			lastRecord = record
			continue
		}

		s.decoder.Reset()
		pushReq, err := s.decoder.Decode(record.Value)
		if err != nil {
			metricRecordsDropped.WithLabelValues(tenant, droppedRecordReasonDecodingFailed).Inc()
			level.Error(s.logger).Log("msg", "failed to decoded record", "tenant", tenant, "err", err)
			span.RecordError(err)
			lastRecord = record
			continue
		}

		// Get or create tenant instance
		inst, err := s.getOrCreateInstance(tenant)
		if err != nil {
			metricRecordsDropped.WithLabelValues(tenant, droppedRecordReasonInstanceNotFound).Inc()
			level.Error(s.logger).Log("msg", "failed to get instance for tenant", "tenant", tenant, "err", err)
			span.RecordError(err)
			lastRecord = record
			continue
		}

		// Push data to tenant instance
		inst.pushBytes(ctx, record.Timestamp, pushReq)

		metricRecordsProcessed.WithLabelValues(tenant).Inc()
		recordCount++
		lastRecord = record
	}

	span.SetAttributes(attribute.Int("records_count", recordCount))

	if lastRecord == nil {
		return nil, nil
	}