fix: Add fallback address and Rider prefix detection for MassTransit (#3519)

GetQueueData now accepts an optional fallback URI (DestinationAddress) for
when SourceAddress yields Unknown — needed for Kafka Rider where the
consume context SourceAddress is the bus endpoint with no queue info.

Rider path prefixes (/kafka/, /event-hub/) are detected regardless of
the bus transport scheme, so Kafka topics are correctly identified as
Topic destination type whether the bus uses RabbitMQ, InMemory, or any
other transport.

Author: tippmar-nr

src/Agent/NewRelic/Agent/Extensions/NewRelic.Agent.Extensions/Helpers/MassTransit.cs

@@ -15,7 +15,21 @@ public class MassTransitQueueData
 
 public class MassTransitHelpers
 {
-    public static MassTransitQueueData GetQueueData(Uri sourceAddress)
+    public static MassTransitQueueData GetQueueData(Uri sourceAddress, Uri fallbackAddress = null)
+    {
+        var data = GetQueueDataFromUri(sourceAddress);
+
+        // If the primary address didn't yield a meaningful name, try the fallback.
+        // This handles transports like Kafka Rider where SourceAddress may be the
+        // bus endpoint (e.g. loopback://localhost/) while DestinationAddress contains
+        // the actual topic/queue information.
+        if (data.QueueName == "Unknown" && fallbackAddress != null)
+            data = GetQueueDataFromUri(fallbackAddress);
+
+        return data;
+    }
+
+    private static MassTransitQueueData GetQueueDataFromUri(Uri sourceAddress)
     {
         var data = new MassTransitQueueData();
 
@@ -41,6 +55,11 @@ public static MassTransitQueueData GetQueueData(Uri sourceAddress)
                 return data;
         }
 
+        // MassTransit Rider embeds path prefixes (e.g. /kafka/, /event-hub/) regardless
+        // of the bus transport scheme. Check for these before scheme-specific parsing.
+        if (TryParseRiderPrefix(sourceAddress, data))
+            return data;
+
         // Transport-specific parsing by URI scheme
         switch (scheme)
         {
@@ -86,6 +105,30 @@ public static MassTransitQueueData GetQueueData(Uri sourceAddress)
         return data;
     }
 
+    private static bool TryParseRiderPrefix(Uri sourceAddress, MassTransitQueueData data)
+    {
+        // MassTransit Rider transports (Kafka, Event Hubs) embed a path prefix
+        // in the URI regardless of the underlying bus transport scheme.
+        // e.g. loopback://localhost/kafka/{topic}, rabbitmq://host/kafka/{topic}
+        var path = sourceAddress.AbsolutePath.TrimStart('/');
+
+        if (path.StartsWith("kafka/", StringComparison.OrdinalIgnoreCase))
+        {
+            data.QueueName = path.Substring("kafka/".Length).TrimEnd('/');
+            data.DestinationType = MessageBrokerDestinationType.Topic;
+            return true;
+        }
+
+        if (path.StartsWith("event-hub/", StringComparison.OrdinalIgnoreCase))
+        {
+            data.QueueName = path.Substring("event-hub/".Length).TrimEnd('/');
+            data.DestinationType = MessageBrokerDestinationType.Topic;
+            return true;
+        }
+
+        return false;
+    }
+
     private static void ParseRabbitMqUri(Uri sourceAddress, MassTransitQueueData data)
     {
         // RabbitMQ uses underscore-delimited names:

src/Agent/NewRelic/Agent/Extensions/Providers/Wrapper/MassTransit/NewRelicFilter.cs

@@ -40,7 +40,7 @@ public async Task Send(ConsumeContext context, IPipe<ConsumeContext> next)
 
         var mc = new MethodCall(_consumeMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CreateTransaction(
             destinationType: queueData.DestinationType,
@@ -86,7 +86,7 @@ public async Task Send(PublishContext context, IPipe<PublishContext> next)
 
         var mc = new MethodCall(_publishMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CurrentTransaction;
         InsertDistributedTraceHeaders(context.Headers, transaction);
@@ -103,7 +103,7 @@ public async Task Send(SendContext context, IPipe<SendContext> next)
 
         var mc = new MethodCall(_sendMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CurrentTransaction;
         InsertDistributedTraceHeaders(context.Headers, transaction);

src/Agent/NewRelic/Agent/Extensions/Providers/Wrapper/MassTransitLegacy/NewRelicFilter.cs

@@ -41,7 +41,7 @@ public async Task Send(ConsumeContext context, IPipe<ConsumeContext> next)
 
         var mc = new MethodCall(_consumeMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CreateTransaction(
             destinationType: queueData.DestinationType,
@@ -87,7 +87,7 @@ public async Task Send(PublishContext context, IPipe<PublishContext> next)
 
         var mc = new MethodCall(_publishMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CurrentTransaction;
         InsertDistributedTraceHeaders(context.Headers, transaction);
@@ -104,7 +104,7 @@ public async Task Send(SendContext context, IPipe<SendContext> next)
 
         var mc = new MethodCall(_sendMethod, context, default(string[]), true);
 
-        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress);
+        var queueData = MassTransitHelpers.GetQueueData(context.SourceAddress, context.DestinationAddress);
 
         var transaction = _agent.CurrentTransaction;
         InsertDistributedTraceHeaders(context.Headers, transaction);