Question about shutdown the queuedRetrySender

[ralphgj]

Hi! We have a question about shutdown the queuedRetrySender.

In https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/queued_retry_inmemory.go#L147 comments, it seems will drain the queue before shutdown the queuedRetrySender.

func (qrs *queuedRetrySender) shutdown() {
	// Cleanup queue metrics reporting
	if qrs.cfg.Enabled {
		_ = globalInstruments.queueSize.UpsertEntry(func() int64 {
			return int64(0)
		}, metricdata.NewLabelValue(qrs.fullName))
	}

	// First Stop the retry goroutines, so that unblocks the queue numWorkers.
	close(qrs.retryStopCh)

	// Stop the queued sender, this will drain the queue and will call the retry (which is stopped) that will only
	// try once every request.
	if qrs.queue != nil {
		qrs.queue.Stop()
	}
}

But in https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/internal/bounded_memory_queue.go#L83, while the select run into the case <-q.stopCh, it will return and then the items in queue will not be consumed by itemConsumer. Are we understand correctly or missing some key points？

func (q *boundedMemoryQueue) StartConsumers(num int, callback func(item interface{})) {
	factory := func() consumer {
		return consumerFunc(callback)
	}
	q.workers = num
	q.factory = factory
	var startWG sync.WaitGroup
	for i := 0; i < q.workers; i++ {
		q.stopWG.Add(1)
		startWG.Add(1)
		go func() {
			startWG.Done()
			defer q.stopWG.Done()
			itemConsumer := q.factory()
			queue := *q.items
			for {
				select {
				case item, ok := <-queue:
					if ok {
						q.size.Sub(1)
						itemConsumer.consume(item)
					} else {
						// channel closed, finish worker
						return
					}
				case <-q.stopCh:
					// the whole queue is closing, finish worker
					return
				}
			}
		}()
	}
	startWG.Wait()
}

[pmm-sumo]

Quickly looking at the code it seems the comment on draining is not accurate. I think that on shutdown the items that were left in memory and not picked up by the workers will be essentially dropped. I think it's worth opening an issue on that

BTW, this problem is largely mitigated by using persistent queue (experimental). The unprocessed items will be picked up after collector restart

[ralphgj]

Thanks. I will open an issue to discuss this question.

BTW, this problem is largely mitigated by using persistent queue (experimental). The unprocessed items will be picked up after collector restart

The collector is deployed in k8s pods, and then the pods will be destroyed when the collector restart. And also it is acceptable that loss some trace data when redeploy the collector or other situations. So we prefer to use the in-memory queue.