Fix race conditions in otelsarama.WrapAsyncProducer (#755)

* Fix possible race condition in WrapAsyncProducer

* Add changelog

Co-authored-by: Tyler Yahn <MrAlias@users.noreply.github.com>

CHANGELOG.md

@@ -8,6 +8,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## [Unreleased]
 
+### Fixed
+
+- Fix race conditions in `otelsarama.WrapAsyncProducer`.
+  The `messaging.message_id` and `messaging.kafka.partition` attributes are now not set if a message was not processed. (#755)
+
 ## [0.21.0] - 2021-06-18
 
 ### Fixed

instrumentation/github.com/Shopify/sarama/otelsarama/producer.go

@@ -152,7 +152,7 @@ func WrapAsyncProducer(saramaConfig *sarama.Config, p sarama.AsyncProducer, opts
 		// cannot be closed.
 		defer func() {
 			for _, mc := range producerMessageContexts {
-				finishProducerSpan(mc.span, 0, 0, nil)
+				mc.span.End()
 			}
 		}()
 		defer close(wrapped.successes)
@@ -194,7 +194,7 @@ func WrapAsyncProducer(saramaConfig *sarama.Config, p sarama.AsyncProducer, opts
 					// If returning successes isn't enabled, we just finish the
 					// span right away because there's no way to know when it will
 					// be done.
-					finishProducerSpan(span, msg.Partition, msg.Offset, nil)
+					span.End()
 				}
 			case msg, ok := <-p.Successes():
 				if !ok {

instrumentation/github.com/Shopify/sarama/otelsarama/producer_test.go

@@ -188,8 +188,6 @@ func TestWrapAsyncProducer(t *testing.T) {
 					semconv.MessagingSystemKey.String("kafka"),
 					semconv.MessagingDestinationKindTopic,
 					semconv.MessagingDestinationKey.String(topic),
-					semconv.MessagingMessageIDKey.String("0"),
-					kafkaPartitionKey.Int64(0),
 				},
 				parentSpanID: oteltrace.SpanID{0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x2},
 				kind:         oteltrace.SpanKindProducer,
@@ -199,8 +197,6 @@ func TestWrapAsyncProducer(t *testing.T) {
 					semconv.MessagingSystemKey.String("kafka"),
 					semconv.MessagingDestinationKindTopic,
 					semconv.MessagingDestinationKey.String(topic),
-					semconv.MessagingMessageIDKey.String("0"),
-					kafkaPartitionKey.Int64(0),
 				},
 				kind: oteltrace.SpanKindProducer,
 			},