Merge pull request #2901 from newrelic/camel

Add additional coverage for Camel Async Processors

Author: obenkenobi

instrumentation/apache-camel-3.9.0/README.md

@@ -0,0 +1,19 @@
+# New Relic Instrumentation: Apache Camel 3.9.0+
+
+Provides distributed tracing and transaction monitoring for applications using [Apache Camel](https://camel.apache.org/) 3.9.0 and above.
+
+## What Is Instrumented
+
+This module hooks into the Apache Camel lifecycle to create and propagate New Relic transactions across Camel routes.
+
+## Configuration
+
+One agent configuration value influences Kafka batch behaviour:
+
+| Config Key | Default | Description                                                                                           |
+|---|---|-------------------------------------------------------------------------------------------------------|
+| `kafka.spans.distributed_trace.consume_many.enabled` | `false` | When `true`, distributed trace headers are accepted from the first exchange in a batch Kafka consume. |
+
+## Notes
+
+- Transactions will not start for certain routing steps (e.g. `direct:`, `seda:`) and/or logging to prevent noise in telemetry.

instrumentation/apache-camel-3.9.0/src/main/java/org/apache/camel/AsyncProcessor_Instrumentation.java

@@ -11,15 +11,23 @@
 import com.newrelic.api.agent.weaver.MatchType;
 import com.newrelic.api.agent.weaver.Weave;
 import com.newrelic.api.agent.weaver.Weaver;
+import com.nr.instrumentation.apache.camel.CamelUtil;
+import com.nr.instrumentation.apache.camel.processors.ExchangeProcessor;
 
 @Weave(originalName = "org.apache.camel.AsyncProcessor", type = MatchType.Interface)
 public class AsyncProcessor_Instrumentation {
 
     @Trace(async = true, excludeFromTransactionTrace = true)
     public boolean process(Exchange exchange, AsyncCallback callback) {
         if (exchange instanceof Exchange_Instrumentation) {
-            if (((Exchange_Instrumentation) exchange).token != null) {
-                ((Exchange_Instrumentation) exchange).token.link();
+            Exchange_Instrumentation exchangeInstrumentation = (Exchange_Instrumentation) exchange;
+            if (exchangeInstrumentation.token != null) {
+                exchangeInstrumentation.token.link();
+            } else if (exchangeInstrumentation.fromConsumer) {
+                ExchangeProcessor exchangeProcessor = CamelUtil.getExchangeProcessor(CamelUtil.getEndpoint(exchange));
+                if (exchangeProcessor != null && exchangeProcessor.shouldStartTransaction()) {
+                    CamelUtil.startTxn(exchange, exchangeProcessor);
+                }
             }
         }
 

instrumentation/apache-camel-3.9.0/src/main/java/org/apache/camel/Processor_Instrumentation.java

@@ -30,8 +30,6 @@ public void process(Exchange exchange) {
                 }
             }
         }
-
-
         Weaver.callOriginal();
     }
 }