From 20425c3aa5039899ed73ca8ffbe17062a73285cb Mon Sep 17 00:00:00 2001
From: Jingguo Yao <yaojingguo@gmail.com>
Date: Sun, 7 Mar 2021 21:48:46 +0800
Subject: [PATCH] Add test cases for kafka-plugin (#6503)

1. Add plugin test for KafkaProducer.send(record).
2. Add exception checking test cases for CallbackInterceptor.
---
 .../plugin/kafka/CallbackInterceptorTest.java |  9 +++++++++
 .../kafka-scenario/config/expectedData.yaml   | 19 +++++++++++++++++++
 .../kafka/controller/CaseController.java      | 17 +++++++++++++++++
 3 files changed, 45 insertions(+)

diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
index 78c36c2b7..4707edb67 100644
--- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
+++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
@@ -152,6 +152,15 @@ public class CallbackInterceptorTest {
         cache.setCallback(callbackAdapterInterceptor);
         EnhancedInstance instance = new CallbackInstance(cache);
         callbackInterceptor.beforeMethod(instance, null, arguments, argumentTypes, null);
+        callbackInterceptor.afterMethod(instance, null, arguments, argumentTypes, null);
+    }
+
+    @Test
+    public void testCallbackWithNullCallbackCache() throws Throwable {
+        EnhancedInstance instance = new CallbackInstance(null);
+        callbackInterceptor.beforeMethod(instance, null, arguments, argumentTypes, null);
+        callbackInterceptor.afterMethod(instance, null, arguments, argumentTypes, null);
+        callbackInterceptor.handleMethodException(instance, null, arguments, argumentTypes, null);
     }
 
     private void assertCallbackSpanWithException(AbstractTracingSpan span) {
diff --git a/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml b/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml
index 5ffb7f473..8c228c33d 100644
--- a/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml
+++ b/test/plugin/scenarios/kafka-scenario/config/expectedData.yaml
@@ -89,6 +89,21 @@ segmentItems:
         - {key: mq.broker, value: 'kafka-server:9092'}
         - {key: mq.topic, value: test2}
       skipAnalysis: 'false'
+    - operationName: Kafka/test2/Producer
+      operationId: 0
+      parentSpanId: 0
+      spanId: 3
+      spanLayer: MQ
+      startTime: nq 0
+      endTime: nq 0
+      componentId: 40
+      isError: false
+      spanType: Exit
+      peer: kafka-server:9092
+      tags:
+        - {key: mq.broker, value: 'kafka-server:9092'}
+        - {key: mq.topic, value: test2}
+      skipAnalysis: 'false'
     - operationName: /case/kafka-case
       operationId: 0
       parentSpanId: -1
@@ -158,10 +173,14 @@ segmentItems:
           - {key: mq.broker, value: 'kafka-server:9092'}
           - {key: mq.topic, value: test.}
           - {key: transmission.latency, value: not null}
+          - {key: transmission.latency, value: not null}
         refs:
           - {parentEndpoint: /case/kafka-case, networkAddress: 'kafka-server:9092', refType: CrossProcess,
              parentSpanId: 2, parentTraceSegmentId: not null, parentServiceInstance: not
                                                                 null, parentService: 'kafka-scenario', traceId: not null}
+          - {parentEndpoint: /case/kafka-case, networkAddress: 'kafka-server:9092', refType: CrossProcess,
+             parentSpanId: 3, parentTraceSegmentId: not null, parentServiceInstance: not
+                                                                null, parentService: 'kafka-scenario', traceId: not null}
         skipAnalysis: 'false'
   - segmentId: not null
     spans:
diff --git a/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java b/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java
index f434fa4a2..e357582a9 100644
--- a/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java
+++ b/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java
@@ -21,6 +21,8 @@ package test.apache.skywalking.apm.testcase.kafka.controller;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Properties;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.regex.Pattern;
 import javax.annotation.PostConstruct;
@@ -93,6 +95,21 @@ public class CaseController {
                 LOGGER.info("send success metadata={}", metadata);
             };
             producer.send(record2, callback2);
+
+            ProducerRecord<String, String> record3 = new ProducerRecord<String, String>(topicName2, "testKey", Integer.toString(1));
+            final Future<RecordMetadata> result = producer.send(record3);
+            new Thread() {
+                @Override
+                public void run() {
+                    try {
+                        // Since linger.ms is 5ms, 10ms should be enough for Kakfa producer to send the message.
+                        RecordMetadata metadata = result.get(10, TimeUnit.MILLISECONDS);
+                        LOGGER.info("send success metadata={}", metadata);
+                    } catch (Throwable t) {
+                        LOGGER.error("failed to send the message to Kafka", t);
+                    }
+                }
+            }.start();
         }, bootstrapServers);
 
         Thread thread = new ConsumerThread();