kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add test cases for kafka-plugin (#6503) 

1. Add plugin test for KafkaProducer.send(record).
2. Add exception checking test cases for CallbackInterceptor.
This commit is contained in:
Jingguo Yao 2021-03-07 21:48:46 +08:00 committed by GitHub
parent 78487ad983
commit 20425c3aa5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 45 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptorTest.java
View File

@ -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) {

19
test/plugin/scenarios/kafka-scenario/config/expectedData.yaml
View File

@ -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:

17
test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java
View File

@ -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();