Fix issues in otelsarama.WrapAsyncProducer

[pellared]

Why

Usage of current Sarama instrumentation can lead to deadlocks and data races. These issues are not present if the instrumentation is not applied (created tests for deadlock scenarios).

The messages returned by the Errors() channel do not contain original Metadata. (I can make it as a separate PR if needed)

What

1. Fix deadlocks by changing the concurrency design.

   • Use dedicated goroutines for consuming messages and for producing messages. It prevents the deadlocks and also makes the implementation less dependent on Sarama's internal implementation.
   • Use dedicated channels for closing instead of reusing the input channel.
   • Redesign the "spans cleanup" to make it NOT dependant on Sarma's implementation details (which may change). See: Possible race condition in Sarama async producer #435
   • All changes were done in a TDD way, however still there may be some edge cases that I have not thought about.

2. Restore messages Metadata for the wrapped Errors channel.

Important Notice

Kudos to @MadVikingGod for a lot of help and for having concerns with the initial fixes 👍 🥇

Still very careful review is needed.

Example deadlock

Reproducing a sample deadlock:

package main

import (
	"fmt"
	"log"

	"github.com/Shopify/sarama"
	"go.opentelemetry.io/contrib/instrumentation/github.com/Shopify/sarama/otelsarama"
)

/*
	EXECUTE:

docker network create confluent

docker run --rm -d \
	--name zookeeper \
	--network confluent \
	-p 2181:2181 \
	-e ZOOKEEPER_CLIENT_PORT=2181 \
	confluentinc/cp-zookeeper:5.0.0

docker run --rm -d \
	--name kafka \
	--network confluent \
	-p 9092:9092 \
	-e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
	-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
	-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
	-e KAFKA_CREATE_TOPICS=gotest:1:1 \
	-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
	confluentinc/cp-kafka:5.0.0

go run -race .

*/

func main() {
	cfg := sarama.NewConfig()
	cfg.Version = sarama.V0_11_0_0 // minimum version that supports headers which are required for tracing

	producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, cfg)
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	producer = otelsarama.WrapAsyncProducer(cfg, producer)

	msg := &sarama.ProducerMessage{
		Topic: "some-topic",
		Value: sarama.StringEncoder("Hello World"),
	}

	// close the producer
	if err := producer.Close(); err != nil {
		log.Fatalln(err)
	}

	// esnure that WrapAsyncProducer has exited
	if _, ok := <-producer.Successes(); ok {
		log.Fatalln("should not receive anything")
	}

	log.Println("Sending message", msg)

	defer func() {
		if r := recover(); r != nil {
			fmt.Println("RECOVERED FROM MAIN:", r)
			return
		}
	}()

	sarama.PanicHandler = func(r interface{}) { // sarama has a data race for PanicHandler...
		fmt.Println("RECOVERED IN SARAMA:", r)
	}

	producer.Input() <- msg // hangs forever (should panic from HERE - not internaly in sarama)

	log.Println("Message sent", msg)
}

[codecov]

Codecov Report

Merging #754 (e938f1b) into main (d41f1f8) will increase coverage by 0.1%.
The diff coverage is 98.2%.

@@           Coverage Diff           @@
##            main    #754     +/-   ##
=======================================
+ Coverage   78.8%   79.0%   +0.1%     
=======================================
  Files         62      62             
  Lines       2707    2726     +19     
=======================================
+ Hits        2135    2154     +19     
  Misses       440     440             
  Partials     132     132             

Impacted Files	Coverage Δ
...n/github.com/Shopify/sarama/otelsarama/producer.go	95.7% <98.2%> (+0.8%)	⬆️

[pellared]

After great tips from @MadVikingGod it ended almost in a rewrite
Also solves: #755

Careful review is needed...

[pellared]

@Aneurysm9 @MadVikingGod Ready for review

PR changed significantly, will re-review.

[MadVikingGod]

I haven't finished, but so far it looks good. The producers of the channels are the ones closing them, There is overall good hygiene around the close semantics. I will try and get to the tests later but consider this a soft +1.

[MadVikingGod]

LGTM