Verify duplicate baggage in telemetry only captures most recent value

Author: phatboyg

src/MassTransit/Logging/Diagnostics/LogContextActivityExtensions.cs

@@ -217,7 +217,7 @@ public static class LogContextActivityExtensions
 
             activity.Start();
 
-            IList<KeyValuePair<string, string?>>? baggage = null;
+            List<KeyValuePair<string, string?>>? baggage = null;
             foreach (KeyValuePair<string, string?> pair in activity.Baggage)
             {
                 if (pair.Key.Equals(DiagnosticHeaders.Messaging.ConversationId) || pair.Key.Equals(DiagnosticHeaders.CorrelationId))
@@ -226,7 +226,7 @@ public static class LogContextActivityExtensions
                 if (string.IsNullOrWhiteSpace(pair.Value))
                     continue;
 
-                baggage ??= new List<KeyValuePair<string, string?>>();
+                baggage ??= new List<KeyValuePair<string, string?>>(1);
                 baggage.Add(pair);
             }
 

tests/MassTransit.RabbitMqTransport.Tests/OpenTelemetry_Specs.cs

@@ -72,6 +72,108 @@ await Assert.MultipleAsync(async () =>
             });
         }
 
+        [Test]
+        public async Task Should_carry_excess_baggage_with_newtonsoft()
+        {
+            var services = new ServiceCollection();
+            AddTraceListener(services, "order-api");
+
+            await using var provider = services
+                .AddMassTransitTestHarness(x =>
+                {
+                    x.AddConsumer<MonitoredSubmitOrderConsumer>();
+
+                    x.UsingRabbitMq((context, cfg) =>
+                    {
+                        cfg.UseNewtonsoftJsonSerializer();
+
+                        cfg.ConfigureEndpoints(context);
+                    });
+                })
+                .BuildServiceProvider(true);
+
+            var harness = provider.GetTestHarness();
+
+            await harness.Start();
+
+            IRequestClient<SubmitOrder> client = harness.GetRequestClient<SubmitOrder>();
+
+            await harness.Bus.Publish(new PingMessage());
+
+            StartedActivity? activity = LogContext.Current?.StartGenericActivity("api process");
+
+            activity?.Activity.SetBaggage("MyBag", "IsFull");
+            activity?.Activity.SetBaggage("MyBag", "IsEmpty");
+            activity?.Activity.SetBaggage("MyBag", "IsHalfFull");
+
+            Response<OrderSubmitted> response = await client.GetResponse<OrderSubmitted>(new
+            {
+                OrderId = InVar.Id,
+                OrderNumber = "123"
+            });
+
+            activity?.Stop();
+
+            await Assert.MultipleAsync(async () =>
+            {
+                Assert.That(await harness.Sent.Any<OrderSubmitted>(), Is.True);
+
+                Assert.That(await harness.Consumed.Any<SubmitOrder>(), Is.True);
+
+                Assert.That(response.Headers.Get<string>("BaggageValue"), Is.EqualTo("IsHalfFull"));
+            });
+        }
+
+        [Test]
+        public async Task Should_carry_excess_baggage()
+        {
+            var services = new ServiceCollection();
+            AddTraceListener(services, "order-api");
+
+            await using var provider = services
+                .AddMassTransitTestHarness(x =>
+                {
+                    x.AddConsumer<MonitoredSubmitOrderConsumer>();
+
+                    x.UsingRabbitMq((context, cfg) =>
+                    {
+                        cfg.ConfigureEndpoints(context);
+                    });
+                })
+                .BuildServiceProvider(true);
+
+            var harness = provider.GetTestHarness();
+
+            await harness.Start();
+
+            IRequestClient<SubmitOrder> client = harness.GetRequestClient<SubmitOrder>();
+
+            await harness.Bus.Publish(new PingMessage());
+
+            StartedActivity? activity = LogContext.Current?.StartGenericActivity("api process");
+
+            activity?.Activity.SetBaggage("MyBag", "IsFull");
+            activity?.Activity.SetBaggage("MyBag", "IsEmpty");
+            activity?.Activity.SetBaggage("MyBag", "IsHalfFull");
+
+            Response<OrderSubmitted> response = await client.GetResponse<OrderSubmitted>(new
+            {
+                OrderId = InVar.Id,
+                OrderNumber = "123"
+            });
+
+            activity?.Stop();
+
+            await Assert.MultipleAsync(async () =>
+            {
+                Assert.That(await harness.Sent.Any<OrderSubmitted>(), Is.True);
+
+                Assert.That(await harness.Consumed.Any<SubmitOrder>(), Is.True);
+
+                Assert.That(response.Headers.Get<string>("BaggageValue"), Is.EqualTo("IsHalfFull"));
+            });
+        }
+
         [Test]
         public async Task Should_report_telemetry_to_jaeger()
         {
@@ -374,6 +476,8 @@ public Task Consume(ConsumeContext<SubmitOrder> context)
             {
                 var value = System.Diagnostics.Activity.Current?.GetBaggageItem("MyBag");
 
+                LogContext.Debug?.Log("Body: {Body}", context.ReceiveContext.Body.GetString());
+
                 return context.RespondAsync<OrderSubmitted>(context.Message, x =>
                 {
                     x.Headers.Set("BaggageValue", value);