[kafka] do not use message creation context as a parent for consumer spans

CHANGELOG.md

@@ -18,6 +18,10 @@ This component adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.h
 
 ### Changed
 
+- Do not use message creation context as a parent for consumer spans for `Confluent.Kafka`
+  client instrumentation. See the [issue](https://github.com/open-telemetry/opentelemetry-dotnet-instrumentation/issues/3434)
+  for details.
+
 #### Dependency updates
 
 - Following packages updated

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/KafkaInstrumentation.cs

@@ -31,22 +31,11 @@ internal static class KafkaInstrumentation
             ? new[] { new ActivityLink(propagatedContext.Value.ActivityContext) }
             : Array.Empty<ActivityLink>();
 
-        // https://github.com/open-telemetry/oteps/blob/5531cb2cb61df5ff9660d1078613e6c09cb396a8/text/trace/0220-messaging-semantic-conventions-span-structure.md?plain=1#L200
-        // If consumer span would be a root span of a new trace,
-        // use message's creation context as a parent,
-        // in addition to linking to it.
-        var parentContext =
-            Activity.Current is null ?
-            propagatedContext?.ActivityContext ?? default :
-            default;
-
         var activity = Source.StartActivity(
-            spanName,
+            name: spanName,
             kind: ActivityKind.Consumer,
             links: activityLinks,
-            startTime: startTime,
-            parentContext: parentContext,
-            tags: null);
+            startTime: startTime);
 
         if (activity is { IsAllDataRequested: true })
         {

test/IntegrationTests/KafkaTests.cs

@@ -31,7 +31,6 @@ public class KafkaTests : TestHelper
 
     // https://github.com/confluentinc/confluent-kafka-dotnet/blob/07de95ed647af80a0db39ce6a8891a630423b952/src/Confluent.Kafka/Offset.cs#L36C44-L36C44
     private const int InvalidOffset = -1001;
-    private const string TestApplicationInstrumentationScopeName = "TestApplication.Kafka";
 
     public KafkaTests(ITestOutputHelper testOutputHelper)
         : base("Kafka", testOutputHelper)
@@ -49,8 +48,6 @@ public void SubmitsTraces(string packageVersion)
         using var collector = new MockSpansCollector(Output);
         SetExporter(collector);
 
-        collector.Expect(TestApplicationInstrumentationScopeName, span => span.Kind == Span.Types.SpanKind.Internal);
-
         // Failed produce attempts made before topic is created.
         collector.Expect(KafkaInstrumentationScopeName, span => span.Kind == Span.Types.SpanKind.Producer && ValidateResultProcessingProduceExceptionSpan(span, topicName), "Failed Produce attempt with delivery handler set.");
         collector.Expect(KafkaInstrumentationScopeName, span => span.Kind == Span.Types.SpanKind.Producer && ValidateProduceExceptionSpan(span, topicName), "Failed Produce attempt without delivery handler set.");
@@ -173,38 +170,16 @@ private static bool ValidatePropagation(ICollection<MockSpansCollector.Collected
                 !span.Span.Attributes.Single(attr => attr.Key == KafkaMessageTombstoneAttributeName).Value.BoolValue &&
                 span.Span.Status is null);
 
-        var consumerSpansWithLinks = producerSpans
-            .Select(producerSpan =>
+        return producerSpans.Select(
+            producerSpan =>
                 GetMatchingConsumerSpan(collectedSpans, producerSpan.Span, expectedReceiveOperationName))
-            .ToList();
-
-        var manuallyCreatedSpan = collectedSpans.Single(collectedSpan =>
-            collectedSpan.InstrumentationScopeName == TestApplicationInstrumentationScopeName);
-
-        var consumerSpanWithCustomParent = consumerSpansWithLinks.SingleOrDefault(
-            span =>
-                span.Span.TraceId == manuallyCreatedSpan.Span.TraceId &&
-                span.Span.ParentSpanId == manuallyCreatedSpan.Span.SpanId);
-
-        var consumerSpansHaveCreationContextAsParent =
-            consumerSpansWithLinks
-                .Except(new[] { consumerSpanWithCustomParent })
-                .All(consumerSpan =>
-                    ValidateCreationContextAsParent(consumerSpan));
-
-        return consumerSpanWithCustomParent is not null && consumerSpansHaveCreationContextAsParent;
-    }
-
-    private static bool ValidateCreationContextAsParent(MockSpansCollector.Collected? consumerSpan)
-    {
-        return consumerSpan?.Span.TraceId == consumerSpan?.Span.Links[0].TraceId &&
-               consumerSpan?.Span.ParentSpanId == consumerSpan?.Span.Links[0].SpanId;
+            .All(consumerSpan => consumerSpan is not null);
     }
 
-    private static MockSpansCollector.Collected GetMatchingConsumerSpan(ICollection<MockSpansCollector.Collected> collectedSpans, Span producerSpan, string expectedReceiveOperationName)
+    private static MockSpansCollector.Collected? GetMatchingConsumerSpan(ICollection<MockSpansCollector.Collected> collectedSpans, Span producerSpan, string expectedReceiveOperationName)
     {
         return collectedSpans
-            .Single(span =>
+            .SingleOrDefault(span =>
             {
                 var parentLinksCount = span.Span.Links.Count(
                     link =>

test/test-applications/integrations/TestApplication.Kafka/Program.cs

@@ -1,7 +1,6 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-using System.Diagnostics;
 using Confluent.Kafka;
 using Confluent.Kafka.Admin;
 
@@ -12,8 +11,6 @@ internal static class Program
     private const string MessageKey = "testkey";
     private const string BootstrapServers = "localhost:9092";
 
-    private static readonly ActivitySource Source = new("TestApplication.Kafka");
-
     public static async Task<int> Main(string[] args)
     {
         var topicName = args[0];
@@ -76,13 +73,7 @@ public static async Task<int> Main(string[] args)
         consumer.Consume(cts.Token);
         consumer.Consume(cts.Token);
 
-        using (var activity = Source.StartActivity("ManuallyStarted"))
-        {
-            consumer.Consume(cts.Token);
-        }
-
-        // Additional attempt that returns no message.
-        consumer.Consume(TimeSpan.FromSeconds(5));
+        consumer.Consume(cts.Token);
 
         // Produce a tombstone.
         producer.Produce(topicName, new Message<string, string> { Key = MessageKey, Value = null! });

test/test-applications/integrations/TestApplication.Kafka/Properties/launchSettings.json

@@ -17,8 +17,7 @@
         "OTEL_DOTNET_AUTO_HOME": "$(SolutionDir)bin\\tracer-home",
         "OTEL_SERVICE_NAME": "TestApplication.Kafka",
         "OTEL_LOG_LEVEL": "debug",
-        "OTEL_DOTNET_AUTO_TRACES_CONSOLE_EXPORTER_ENABLED": "true",
-        "OTEL_DOTNET_AUTO_TRACES_ADDITIONAL_SOURCES": "TestApplication.Kafka"
+        "OTEL_DOTNET_AUTO_TRACES_CONSOLE_EXPORTER_ENABLED": "true"
       }
     }
   }

test/test-applications/integrations/TestApplication.Kafka/TestApplication.Kafka.csproj

@@ -1,6 +1,5 @@
 <Project Sdk="Microsoft.NET.Sdk"> 
   <ItemGroup>
     <PackageReference Include="Confluent.Kafka" VersionOverride="$(LibraryVersion)" />
-    <PackageReference Include="System.Diagnostics.DiagnosticSource" />
   </ItemGroup>
 </Project>