getMethodsMatcher() {
+ return named(SUBSCRIBE_METHOD).and(takesArgumentWithType(1, SUBSCRIBE_INTERCEPT_TYPE));
+ }
+
+ @Override public String getMethodsInterceptor() {
+ return "org.apache.skywalking.apm.plugin.kafka.v11.SubscribeMethodInterceptor";
+ }
+
+ @Override public boolean isOverrideArgs() {
+ return false;
+ }
+ }
+ };
+ }
+
+ @Override protected ClassMatch enhanceClass() {
+ return byName(ENHANCE_CLASS);
+ }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/KafkaProducerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/KafkaProducerInstrumentation.java
new file mode 100644
index 000000000..54932a8af
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/KafkaProducerInstrumentation.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v1.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.kafka.v1.KafkaProducerInterceptor;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link KafkaProducerInstrumentation} define that {@link KafkaProducerInterceptor}
+ * intercept the method send in the class org.apache.kafka.clients.producer.KafkaProducer.
+ * Here is the intercept process steps.
+ *
+ *
+ *
+ * 1. Record the broker address when the client create the org.apache.kafka.clients.producer.KafkaProducer
+ * instance
+ * 2. Create the exit span when the client invoke send method
+ * 3. Inject the context to {@link org.apache.kafka.clients.producer.ProducerRecord#headers}
+ * 3. Stop the exit span when send method finished.
+ *
+ *
+ * @author zhang xin
+ */
+public class KafkaProducerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+ public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.KafkaProducerInterceptor";
+ public static final String ENHANCE_CLASS = "org.apache.kafka.clients.producer.KafkaProducer";
+ public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ProducerConstructorInterceptor";
+ public static final String CONSTRUCTOR_INTERCEPTOR_FLAG = "org.apache.kafka.clients.producer.ProducerConfig";
+
+ @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+ return new ConstructorInterceptPoint[] {
+ new ConstructorInterceptPoint() {
+ @Override public ElementMatcher getConstructorMatcher() {
+ return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPTOR_FLAG);
+ }
+
+ @Override public String getConstructorInterceptor() {
+ return CONSTRUCTOR_INTERCEPTOR_CLASS;
+ }
+ }
+ };
+ }
+
+ @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+ return new InstanceMethodsInterceptPoint[] {
+ new InstanceMethodsInterceptPoint() {
+ @Override public ElementMatcher getMethodsMatcher() {
+ return named("doSend");
+ }
+
+ @Override public String getMethodsInterceptor() {
+ return INTERCEPTOR_CLASS;
+ }
+
+ @Override public boolean isOverrideArgs() {
+ return false;
+ }
+ }
+ };
+ }
+
+ @Override protected ClassMatch enhanceClass() {
+ return byName(ENHANCE_CLASS);
+ }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/ProducerRecordInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/ProducerRecordInstrumentation.java
new file mode 100644
index 000000000..fe4d8b203
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v1/define/ProducerRecordInstrumentation.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v1.define;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.kafka.v1.ProducerRecordConstructorInterceptor;
+
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
+
+/**
+ * {@link ProducerRecordInstrumentation} define that {@link ProducerRecordConstructorInterceptor}
+ * intercept the constructor in the class org.apache.kafka.clients.producer.ProducerRecord for record the
+ * topic name and propagate the Context of trace.
+ *
+ * @author zhang xin
+ * @see org.apache.skywalking.apm.plugin.kafka.v1.define.KafkaProducerInstrumentation
+ */
+public class ProducerRecordInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+
+ public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ProducerRecordConstructorInterceptor";
+ public static final String ENHANCE_CLASS = "org.apache.kafka.clients.producer.ProducerRecord";
+
+ @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+ return new ConstructorInterceptPoint[] {
+ new ConstructorInterceptPoint() {
+ @Override public ElementMatcher getConstructorMatcher() {
+ return takesArguments(6);
+ }
+
+ @Override public String getConstructorInterceptor() {
+ return CONSTRUCTOR_INTERCEPTOR_CLASS;
+ }
+ }
+ };
+ }
+
+ @Override protected InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+ return new InstanceMethodsInterceptPoint[0];
+ }
+
+ @Override protected ClassMatch enhanceClass() {
+ return byName(ENHANCE_CLASS);
+ }
+}
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/resources/skywalking-plugin.def
new file mode 100644
index 000000000..a6e5972a0
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/main/resources/skywalking-plugin.def
@@ -0,0 +1,4 @@
+kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v1.define.CallbackInstrumentation
+kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v1.define.KafkaConsumerInstrumentation
+kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v1.define.KafkaProducerInstrumentation
+kafka-0.11.x=org.apache.skywalking.apm.plugin.kafka.v1.define.ProducerRecordInstrumentation
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java
new file mode 100644
index 000000000..17ddaf443
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/CallbackInterceptorTest.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.List;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.skywalking.apm.agent.core.context.MockContextSnapshot;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegmentRef;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SpanHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentRefAssert;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.SpanAssert;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.apache.skywalking.apm.plugin.kafka.v1.CallbackInterceptor;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+@PrepareForTest({RecordMetadata.class})
+public class CallbackInterceptorTest {
+
+ @SegmentStoragePoint
+ private SegmentStorage segmentStorage;
+
+ @Rule
+ public AgentServiceRule serviceRule = new AgentServiceRule();
+
+ @Mock
+ private RecordMetadata recordMetadata;
+
+ private CallbackInterceptor callbackInterceptor;
+
+ private Object[] arguments;
+ private Object[] argumentsWithException;
+ private Class[] argumentTypes;
+
+ private EnhancedInstance callBackInstance = new EnhancedInstance() {
+ @Override public Object getSkyWalkingDynamicField() {
+ return MockContextSnapshot.INSTANCE.mockContextSnapshot();
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+
+ }
+ };
+
+ @Before
+ public void setUp() {
+ callbackInterceptor = new CallbackInterceptor();
+
+ arguments = new Object[] {
+ recordMetadata, null
+ };
+ argumentsWithException = new Object[] {
+ recordMetadata, new RuntimeException()
+ };
+
+ argumentTypes = new Class[] {
+ RecordMetadata.class, Exception.class
+ };
+
+ }
+
+ @Test
+ public void testCallbackWithoutException() throws Throwable {
+ callbackInterceptor.beforeMethod(callBackInstance, null, arguments, argumentTypes, null);
+ callbackInterceptor.afterMethod(callBackInstance, null, arguments, argumentTypes, null);
+
+ List traceSegments = segmentStorage.getTraceSegments();
+ assertThat(traceSegments.size(), is(1));
+ TraceSegment traceSegment = traceSegments.get(0);
+
+ List abstractSpans = SegmentHelper.getSpans(traceSegment);
+ assertThat(abstractSpans.size(), is(1));
+
+ assertCallbackSpan(abstractSpans.get(0));
+
+ assertCallbackSegmentRef(traceSegment.getRefs());
+ }
+
+ @Test
+ public void testCallbackWithException() throws Throwable {
+ callbackInterceptor.beforeMethod(callBackInstance, null, argumentsWithException, argumentTypes, null);
+ callbackInterceptor.afterMethod(callBackInstance, null, argumentsWithException, argumentTypes, null);
+
+ List traceSegments = segmentStorage.getTraceSegments();
+ assertThat(traceSegments.size(), is(1));
+ TraceSegment traceSegment = traceSegments.get(0);
+
+ List abstractSpans = SegmentHelper.getSpans(traceSegment);
+ assertThat(abstractSpans.size(), is(1));
+
+ assertCallbackSpanWithException(abstractSpans.get(0));
+
+ assertCallbackSegmentRef(traceSegment.getRefs());
+ }
+
+ private void assertCallbackSpanWithException(AbstractTracingSpan span) {
+ assertCallbackSpan(span);
+
+ SpanAssert.assertException(SpanHelper.getLogs(span).get(0), RuntimeException.class);
+ assertThat(SpanHelper.getErrorOccurred(span), is(true));
+ }
+
+ private void assertCallbackSegmentRef(List refs) {
+ assertThat(refs.size(), is(1));
+
+ TraceSegmentRef segmentRef = refs.get(0);
+ SegmentRefAssert.assertSpanId(segmentRef, 1);
+ assertThat(segmentRef.getEntryOperationName(), is("/for-test-entryOperationName"));
+ }
+
+ private void assertCallbackSpan(AbstractTracingSpan span) {
+ assertThat(span.getOperationName(), is("Producer/Callback"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java
new file mode 100644
index 000000000..e6125ba78
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ConsumerConstructorInterceptorTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.plugin.kafka.v1.ConsumerConstructorInterceptor;
+import org.apache.skywalking.apm.plugin.kafka.v1.ConsumerEnhanceRequiredInfo;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ConsumerConstructorInterceptorTest {
+
+ @Mock
+ private ConsumerConfig consumerConfig;
+
+ @Mock
+ private ConsumerConstructorInterceptor constructorInterceptor;
+
+ private EnhancedInstance enhancedInstance = new EnhancedInstance() {
+ private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
+
+ @Override public Object getSkyWalkingDynamicField() {
+ return consumerEnhanceRequiredInfo;
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+ consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value;
+ }
+ };
+
+ @Before
+ public void setUp() {
+ List mockBootstrapServers = new ArrayList();
+ mockBootstrapServers.add("localhost:9092");
+ mockBootstrapServers.add("localhost:19092");
+ when(consumerConfig.getList("bootstrap.servers")).thenReturn(mockBootstrapServers);
+
+ constructorInterceptor = new ConsumerConstructorInterceptor();
+ }
+
+ @Test
+ public void testOnConsumer() {
+ constructorInterceptor.onConstruct(enhancedInstance, new Object[] {consumerConfig});
+ ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)enhancedInstance.getSkyWalkingDynamicField();
+ assertThat(consumerEnhanceRequiredInfo.getBrokerServers(), is("localhost:9092;localhost:19092"));
+ }
+
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java
new file mode 100644
index 000000000..06479bcda
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptorTest.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegmentRef;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.helper.SegmentRefHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.SpanAssert;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.apache.skywalking.apm.plugin.kafka.v1.ConsumerEnhanceRequiredInfo;
+import org.apache.skywalking.apm.plugin.kafka.v1.KafkaConsumerInterceptor;
+import org.hamcrest.MatcherAssert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+public class KafkaConsumerInterceptorTest {
+
+ @SegmentStoragePoint
+ private SegmentStorage segmentStorage;
+
+ @Rule
+ public AgentServiceRule serviceRule = new AgentServiceRule();
+
+ private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo;
+
+ private KafkaConsumerInterceptor consumerInterceptor;
+
+ private EnhancedInstance consumerInstance = new EnhancedInstance() {
+ @Override public Object getSkyWalkingDynamicField() {
+ return consumerEnhanceRequiredInfo;
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+ consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value;
+ }
+ };
+
+ private Map> messages;
+
+ @Before
+ public void setUp() {
+ consumerInterceptor = new KafkaConsumerInterceptor();
+ consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
+
+ List topics = new ArrayList();
+ topics.add("test");
+ topics.add("test-1");
+ consumerEnhanceRequiredInfo.setTopics(topics);
+ List brokers = new ArrayList();
+ brokers.add("localhost:9092");
+ brokers.add("localhost:19092");
+ consumerEnhanceRequiredInfo.setBrokerServers(brokers);
+
+ messages = new HashMap>();
+ TopicPartition topicPartition = new TopicPartition("test", 1);
+ List records = new ArrayList();
+ ConsumerRecord consumerRecord = new ConsumerRecord("test", 1, 0, "1", "1");
+ consumerRecord.headers().add("sw3", "1.234.111|3|1|1|#192.168.1.8:18002|#/portal/|#testEntrySpan|#AQA*#AQA*Et0We0tQNQA*".getBytes());
+ records.add(consumerRecord);
+ messages.put(topicPartition, records);
+ }
+
+ @Test
+ public void testConsumerWithoutMessage() throws Throwable {
+ consumerInterceptor.beforeMethod(consumerInstance, null, new Object[0], new Class[0], null);
+ consumerInterceptor.afterMethod(consumerInstance, null, new Object[0], new Class[0], new HashMap>());
+
+ List traceSegments = segmentStorage.getTraceSegments();
+ assertThat(traceSegments.size(), is(0));
+ }
+
+ @Test
+ public void testConsumerWithMessage() throws Throwable {
+ consumerInterceptor.beforeMethod(consumerInstance, null, new Object[0], new Class[0], null);
+ consumerInterceptor.afterMethod(consumerInstance, null, new Object[0], new Class[0], messages);
+
+ List traceSegments = segmentStorage.getTraceSegments();
+ assertThat(traceSegments.size(), is(1));
+
+ TraceSegment traceSegment = traceSegments.get(0);
+ List refs = traceSegment.getRefs();
+ assertThat(refs.size(), is(1));
+ assertTraceSegmentRef(refs.get(0));
+
+ List spans = SegmentHelper.getSpans(traceSegment);
+ assertThat(spans.size(), is(1));
+ assertConsumerSpan(spans.get(0));
+ }
+
+ private void assertConsumerSpan(AbstractTracingSpan span) {
+ SpanAssert.assertLayer(span, SpanLayer.MQ);
+ SpanAssert.assertComponent(span, KAFKA);
+ SpanAssert.assertTagSize(span, 2);
+ SpanAssert.assertTag(span, 0, "localhost:9092;localhost:19092");
+ SpanAssert.assertTag(span, 1, "test;test-1");
+ }
+
+ private void assertTraceSegmentRef(TraceSegmentRef ref) {
+ MatcherAssert.assertThat(SegmentRefHelper.getEntryApplicationInstanceId(ref), is(1));
+ MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3));
+ MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("1.234.111"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java
new file mode 100644
index 000000000..aff40ca32
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptorTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.List;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.skywalking.apm.agent.core.context.trace.AbstractTracingSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.helper.SegmentHelper;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.SpanAssert;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.apache.skywalking.apm.plugin.kafka.v1.KafkaProducerInterceptor;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockRunnerDelegate(TracingSegmentRunner.class)
+public class KafkaProducerInterceptorTest {
+ @SegmentStoragePoint
+ private SegmentStorage segmentStorage;
+
+ @Rule
+ public AgentServiceRule serviceRule = new AgentServiceRule();
+
+ private KafkaProducerInterceptor producerInterceptor;
+
+ private Object[] arguments;
+ private Class[] argumentType;
+
+ private EnhancedInstance kafkaProducerInstance = new EnhancedInstance() {
+ @Override public Object getSkyWalkingDynamicField() {
+ return "localhost:9092";
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+
+ }
+ };
+
+ private EnhancedInstance messageInstance = new MockProducerMessage();
+
+ private class MockProducerMessage extends ProducerRecord implements EnhancedInstance {
+
+ public MockProducerMessage() {
+ super("test", "");
+ }
+
+ @Override public Object getSkyWalkingDynamicField() {
+ return "test";
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+
+ }
+ }
+
+ @Before
+ public void setUp() {
+ producerInterceptor = new KafkaProducerInterceptor();
+
+ arguments = new Object[] {messageInstance, null};
+ argumentType = new Class[] {ProducerRecord.class};
+ }
+
+ @Test
+ public void testSendMessage() throws Throwable {
+ producerInterceptor.beforeMethod(kafkaProducerInstance, null, arguments, argumentType, null);
+ producerInterceptor.afterMethod(kafkaProducerInstance, null, arguments, argumentType, null);
+
+ List traceSegmentList = segmentStorage.getTraceSegments();
+ assertThat(traceSegmentList.size(), is(1));
+
+ TraceSegment segment = traceSegmentList.get(0);
+ List spans = SegmentHelper.getSpans(segment);
+ assertThat(spans.size(), is(1));
+
+ assertMessageSpan(spans.get(0));
+ }
+
+ private void assertMessageSpan(AbstractTracingSpan span) {
+ SpanAssert.assertTag(span, 0, "localhost:9092");
+ SpanAssert.assertTag(span, 1, "test");
+ SpanAssert.assertComponent(span, KAFKA);
+ SpanAssert.assertLayer(span, SpanLayer.MQ);
+ assertThat(span.getOperationName(), is("Kafka/test/Producer"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java
new file mode 100644
index 000000000..6eee953f5
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptorTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.plugin.kafka.v1.ProducerConstructorInterceptor;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ProducerConstructorInterceptorTest {
+ @Mock
+ private ProducerConfig producerConfig;
+
+ @Mock
+ private ProducerConstructorInterceptor constructorInterceptor;
+
+ private EnhancedInstance enhancedInstance = new EnhancedInstance() {
+ private String brokerServers;
+
+ @Override public Object getSkyWalkingDynamicField() {
+ return brokerServers;
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+ brokerServers = (String)value;
+ }
+ };
+
+ @Before
+ public void setUp() {
+ List mockBootstrapServers = new ArrayList();
+ mockBootstrapServers.add("localhost:9092");
+ mockBootstrapServers.add("localhost:19092");
+ when(producerConfig.getList("bootstrap.servers")).thenReturn(mockBootstrapServers);
+ constructorInterceptor = new ProducerConstructorInterceptor();
+ }
+
+ @Test
+ public void testOnConsumer() {
+ constructorInterceptor.onConstruct(enhancedInstance, new Object[] {producerConfig});
+ assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("localhost:9092;localhost:19092"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java
new file mode 100644
index 000000000..5a1a9d677
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptorTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.plugin.kafka.v1.ProducerRecordConstructorInterceptor;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ProducerRecordConstructorInterceptorTest {
+
+ @Mock
+ private ProducerRecordConstructorInterceptor constructorInterceptor;
+
+ private EnhancedInstance enhancedInstance = new EnhancedInstance() {
+ private String brokerServers;
+
+ @Override public Object getSkyWalkingDynamicField() {
+ return brokerServers;
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+ brokerServers = (String)value;
+ }
+ };
+
+ @Before
+ public void setUp() {
+ constructorInterceptor = new ProducerRecordConstructorInterceptor();
+ }
+
+ @Test
+ public void testOnConsumer() {
+ constructorInterceptor.onConstruct(enhancedInstance, new Object[] {"test"});
+ assertThat(enhancedInstance.getSkyWalkingDynamicField().toString(), is("test"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java
new file mode 100644
index 000000000..66a084293
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-v1-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/v11/SubscribeMethodInterceptorTest.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.kafka.v11;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.plugin.kafka.v1.ConsumerEnhanceRequiredInfo;
+import org.apache.skywalking.apm.plugin.kafka.v1.SubscribeMethodInterceptor;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SubscribeMethodInterceptorTest {
+
+ @Mock
+ private SubscribeMethodInterceptor constructorInterceptor;
+
+ private List mockTopics = new ArrayList();
+
+ private EnhancedInstance enhancedInstance = new EnhancedInstance() {
+ ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo = new ConsumerEnhanceRequiredInfo();
+
+ @Override public Object getSkyWalkingDynamicField() {
+ return consumerEnhanceRequiredInfo;
+ }
+
+ @Override public void setSkyWalkingDynamicField(Object value) {
+ this.consumerEnhanceRequiredInfo = (ConsumerEnhanceRequiredInfo)value;
+ }
+ };
+
+ @Before
+ public void setUp() {
+ mockTopics.add("test");
+ mockTopics.add("test-1");
+ constructorInterceptor = new SubscribeMethodInterceptor();
+ }
+
+ @Test
+ public void testOnConsumer() throws Throwable {
+ constructorInterceptor.beforeMethod(enhancedInstance, null, new Object[] {mockTopics}, new Class[] {Collection.class}, null);
+ ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)enhancedInstance.getSkyWalkingDynamicField();
+ assertThat(requiredInfo.getTopics(), is("test;test-1"));
+ }
+}
\ No newline at end of file
diff --git a/apm-sniffer/apm-sdk-plugin/pom.xml b/apm-sniffer/apm-sdk-plugin/pom.xml
index dba83ca90..095740230 100644
--- a/apm-sniffer/apm-sdk-plugin/pom.xml
+++ b/apm-sniffer/apm-sdk-plugin/pom.xml
@@ -53,6 +53,7 @@
elastic-job-2.x-plugin
mongodb-2.x-plugin
httpasyncclient-4.x-plugin
+ kafka-v1-plugin
pom
diff --git a/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java b/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java
new file mode 100644
index 000000000..afc4f8a51
--- /dev/null
+++ b/apm-sniffer/apm-test-tools/src/main/java/org/apache/skywalking/apm/agent/core/context/MockContextSnapshot.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.agent.core.context;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.skywalking.apm.agent.core.context.ids.DistributedTraceId;
+import org.apache.skywalking.apm.agent.core.context.ids.ID;
+import org.apache.skywalking.apm.agent.core.context.ids.NewDistributedTraceId;
+
+public enum MockContextSnapshot {
+ INSTANCE;
+
+ private ContextSnapshot contextSnapshot;
+
+ MockContextSnapshot() {
+ List distributedTraceIds = new ArrayList();
+ distributedTraceIds.add(new NewDistributedTraceId());
+
+ contextSnapshot = new ContextSnapshot(new ID(1, 2, 3), 1, distributedTraceIds);
+ contextSnapshot.setEntryApplicationInstanceId(1);
+ contextSnapshot.setEntryOperationId(0);
+ contextSnapshot.setEntryOperationName("/for-test-entryOperationName");
+ contextSnapshot.setParentOperationId(0);
+ contextSnapshot.setParentOperationName("/for-test-parentOperationName");
+ }
+
+ public ContextSnapshot mockContextSnapshot() {
+ return contextSnapshot;
+ }
+}