Produce() from kafka.Producer will always return no errors if cfg.Sync=true

[endorama]

Produce() method of kafka.Producer has this signature:

apm-queue/kafka/producer.go

Line 210 in a508dbd

func (p *Producer) Produce(ctx context.Context, rs ...apmqueue.Record) error {

It also has 2 behaviors: async and sync production. This behavior is triggered by the cfg.Sync config option.

In the async production case the errors occurring are passed over to the cfg.ProduceCallback if specified. This is correct.

In the sync production case errors occurring are not collected in any way and will never be returned to the caller. Worse: the caller will never know an error happened.

This is the relevant code where we produce to kafka:

apm-queue/kafka/producer.go

Lines 243 to 258 in a508dbd

	p.client.Produce(ctx, kgoRecord, func(r *kgo.Record, err error) {
	defer wg.Done()
	// kotel already marks spans as errors. No need to handle it here.
	if err != nil {
	p.cfg.Logger.Error("failed producing message",
	zap.Error(err),
	zap.String("topic", strings.TrimPrefix(r.Topic, namespacePrefix)),
	zap.Int64("offset", r.Offset),
	zap.Int32("partition", r.Partition),
	zap.Any("headers", headers),
	)
	}
	if p.cfg.ProduceCallback != nil {
	p.cfg.ProduceCallback(r, err)
	}
	})

I'm not sure this behavior is intended and I would consider it a bug. The solution could be to collect errors with errors.Join and return the resulting error on line 261

apm-queue/kafka/producer.go

Lines 260 to 264 in a508dbd

	if p.cfg.Sync {
	wg.Wait()
	}
	return nil
	}

This way on cfg.Sync=true errors would be collected and returned. On cfg.Sync=false the code would not wait thus returning nil keeping the current behavior.

[simitt]

@endorama is my understanding correct that this poses a problem for non-production environment?