Add null check to kafka

dd-java-agent/instrumentation/aws-sdk/src/main/java/datadog/trace/instrumentation/aws/AWSClientInstrumentation.java

@@ -48,6 +48,7 @@ public AgentBuilder apply(final AgentBuilder agentBuilder) {
   }
 
   public static class AWSClientAdvice {
+    // Since we're instrumenting the constructor, we can't add onThrowable.
     @Advice.OnMethodExit(suppress = Throwable.class)
     public static void addHandler(
         @Advice.FieldValue("requestHandler2s") final List<RequestHandler2> handlers) {

dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/FutureInstrumentation.java

@@ -61,8 +61,7 @@ public final class FutureInstrumentation extends Instrumenter.Configurable {
       "com.google.common.util.concurrent.AbstractFuture$TrustedFuture",
       "com.google.common.util.concurrent.AbstractFuture"
     };
-    WHITELISTED_FUTURES =
-        Collections.unmodifiableSet(new HashSet<String>(Arrays.asList(whitelist)));
+    WHITELISTED_FUTURES = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(whitelist)));
   }
 
   public FutureInstrumentation() {
@@ -81,7 +80,7 @@ public AgentBuilder apply(final AgentBuilder agentBuilder) {
         .and(
             new ElementMatcher<TypeDescription>() {
               @Override
-              public boolean matches(TypeDescription target) {
+              public boolean matches(final TypeDescription target) {
                 final boolean whitelisted = WHITELISTED_FUTURES.contains(target.getName());
                 if (!whitelisted) {
                   log.debug("Skipping future instrumentation for {}", target.getName());
@@ -101,13 +100,13 @@ public boolean matches(TypeDescription target) {
 
   public static class CanceledFutureAdvice {
     @Advice.OnMethodEnter(suppress = Throwable.class)
-    public static DatadogWrapper findWrapper(@Advice.This Future<?> future) {
+    public static DatadogWrapper findWrapper(@Advice.This final Future<?> future) {
       return ConcurrentUtils.getDatadogWrapper(future);
     }
 
     @Advice.OnMethodExit(suppress = Throwable.class)
     public static void abortTracing(
-        @Advice.Enter final DatadogWrapper wrapper, @Advice.Return boolean canceled) {
+        @Advice.Enter final DatadogWrapper wrapper, @Advice.Return final boolean canceled) {
       if (canceled && null != wrapper) {
         wrapper.cancel();
       }

dd-java-agent/instrumentation/jdbc/src/main/java/datadog/trace/instrumentation/jdbc/ConnectionInstrumentation.java

@@ -61,6 +61,7 @@ public static int constructorEnter() {
       return CallDepthThreadLocalMap.get(Connection.class).incrementCallDepth();
     }
 
+    // Since we're instrumenting the constructor, we can't add onThrowable.
     @Advice.OnMethodExit(suppress = Throwable.class)
     public static void constructorExit(
         @Advice.Enter final int depth, @Advice.This final Connection connection)

dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaConsumerInstrumentation.java

@@ -71,17 +71,21 @@ public static class IterableAdvice {
 
     @Advice.OnMethodExit(suppress = Throwable.class)
     public static void wrap(@Advice.Return(readOnly = false) Iterable<ConsumerRecord> iterable) {
-      iterable = new TracingIterable(iterable, "kafka.consume", ConsumeScopeAction.INSTANCE);
+      if (iterable != null) {
+        iterable = new TracingIterable(iterable, "kafka.consume", ConsumeScopeAction.INSTANCE);
+      }
     }
   }
 
   public static class IteratorAdvice {
 
     @Advice.OnMethodExit(suppress = Throwable.class)
     public static void wrap(@Advice.Return(readOnly = false) Iterator<ConsumerRecord> iterator) {
-      iterator =
-          new TracingIterable.TracingIterator(
-              iterator, "kafka.consume", ConsumeScopeAction.INSTANCE);
+      if (iterator != null) {
+        iterator =
+            new TracingIterable.TracingIterator(
+                iterator, "kafka.consume", ConsumeScopeAction.INSTANCE);
+      }
     }
   }
 

dd-java-agent/instrumentation/kafka-streams-0.11/src/main/java/datadog/trace/instrumentation/kafka_streams/KafkaStreamsProcessorInstrumentation.java

@@ -112,11 +112,11 @@ public AgentBuilder apply(final AgentBuilder agentBuilder) {
               DDAdvice.create()
                   .advice(
                       isMethod().and(isPublic()).and(named("process")).and(takesArguments(0)),
-                      StartSpanAdvice.class.getName()))
+                      StopSpanAdvice.class.getName()))
           .asDecorator();
     }
 
-    public static class StartSpanAdvice {
+    public static class StopSpanAdvice {
 
       @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
       public static void stopSpan(@Advice.Thrown final Throwable throwable) {

dd-java-agent/instrumentation/sparkjava-2.4/src/main/java/datadog/trace/instrumentation/sparkjava/RoutesInstrumentation.java

@@ -45,7 +45,7 @@ public AgentBuilder apply(final AgentBuilder agentBuilder) {
 
   public static class RoutesAdvice {
 
-    @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+    @Advice.OnMethodExit(suppress = Throwable.class)
     public static void routeMatchEnricher(
         @Advice.Argument(0) final HttpMethod method, @Advice.Return final RouteMatch routeMatch) {
 

dd-java-agent/testing/src/test/java/IBMResourceLevelInstrumentation.java

@@ -14,7 +14,7 @@ public IBMResourceLevelInstrumentation() {
   }
 
   @Override
-  protected AgentBuilder apply(AgentBuilder agentBuilder) {
+  protected AgentBuilder apply(final AgentBuilder agentBuilder) {
     return agentBuilder
         .type(named("com.ibm.as400.resource.ResourceLevel"))
         .transform(DDTransformers.defaultTransformers())

dd-trace-ot/src/main/java/datadog/trace/common/writer/ListWriter.java

@@ -38,7 +38,7 @@ public void waitForTraces(final int number) throws InterruptedException, Timeout
       }
       latches.add(latch);
     }
-    if (!latch.await(5, TimeUnit.SECONDS)) {
+    if (!latch.await(20, TimeUnit.SECONDS)) {
       throw new TimeoutException("Timeout waiting for " + number + " trace(s).");
     }
   }