Method fetchMessages

backend/pkg/console/list_messages.go:828–932

FetchMessages is in charge of fulfilling the topic consume request. This is tricky in many cases, often due to the fact that we can't consume backwards, but we offer users to consume the most recent messages.

(ctx context.Context, cl *kgo.Client, progress IListMessagesProgress, consumeReq TopicConsumeRequest)

// in many cases, often due to the fact that we can't consume backwards, but we offer
// users to consume the most recent messages.
func (s *Service) fetchMessages(ctx context.Context, cl *kgo.Client, progress IListMessagesProgress, consumeReq TopicConsumeRequest) error {
	// 1. Assign partitions with right start offsets and create client
	partitionOffsets := make(map[string]map[int32]kgo.Offset)
	partitionOffsets[consumeReq.TopicName] = make(map[int32]kgo.Offset)
	for _, req := range consumeReq.Partitions {
		offset := kgo.NewOffset().At(req.StartOffset)
		partitionOffsets[consumeReq.TopicName][req.PartitionID] = offset
	}

	opts := append(cl.Opts(), kgo.ConsumePartitions(partitionOffsets))
	client, err := kgo.NewClient(opts...)
	if err != nil {
		return fmt.Errorf("failed to create new kafka client: %w", err)
	}
	defer client.Close()

	// 2. Create consumer workers
	// Reduced from 100 to 20 to limit memory usage in serverless environments
	// With large records (up to 1MB), 100 records = 1GB+ after deserialization
	jobs := make(chan *kgo.Record, 20)
	resultsCh := make(chan *TopicMessage, 20)
	workerCtx, cancel := context.WithCancelCause(ctx)
	defer cancel(errors.New("worker cancel"))

	wg := sync.WaitGroup{}

	// If we use more than one worker the order of messages in each partition gets lost. Hence we only use it where
	// multiple workers are actually beneficial - for potentially high throughput stream requests.
	workerCount := 1
	if consumeReq.FilterInterpreterCode != "" {
		workerCount = 6
	}
	for i := 0; i < workerCount; i++ {
		// Setup JavaScript interpreter
		isMessageOK, err := s.setupInterpreter(consumeReq.FilterInterpreterCode)
		if err != nil {
			s.logger.ErrorContext(ctx, "failed to setup interpreter", slog.Any("error", err))
			progress.OnError(fmt.Sprintf("failed to setup interpreter: %v", err.Error()))
			return err
		}

		wg.Add(1)
		go s.startMessageWorker(workerCtx, &wg, isMessageOK, jobs, resultsCh,
			consumeReq)
	}
	// Close the results channel once all workers have finished processing jobs and therefore no senders are left anymore
	go func() {
		wg.Wait()
		close(resultsCh)
	}()

	// 3. Start go routine that consumes messages from Kafka and produces these records on the jobs channel so that these
	// can be decoded by our workers.
	go s.consumeKafkaMessages(workerCtx, client, consumeReq, jobs)

	// 4. Receive decoded messages until our request is satisfied. Once that's the case we will cancel the context
	// that propagate to all the launched go routines.
	messageCount := 0