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 dc492895d..197c1bc11 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
servicecomb-plugin
pom
@@ -65,6 +66,9 @@
net.bytebuddy
${shade.package}.${shade.net.bytebuddy.source}
+
+ ${project.build.directory}${sdk.plugin.related.dir}/../../../../packages/skywalking-agent
+ ${agent.package.dest.dir}/plugins
@@ -168,10 +172,10 @@
+ dir="${plugin.dest.dir}"/>
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;
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/pom.xml b/apm-sniffer/optional-plugins/optional-spring-plugins/pom.xml
new file mode 100644
index 000000000..e3b38f27c
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/pom.xml
@@ -0,0 +1,40 @@
+
+
+
+
+
+ optional-plugins
+ org.apache.skywalking
+ 5.0.0-alpha
+
+ 4.0.0
+ pom
+
+ optional-spring-plugins
+
+ ${project.build.directory}/../../../../../packages/skywalking-agent/optional-plugins
+
+
+
+ spring-annotation-plugin
+
+
+
\ No newline at end of file
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/pom.xml b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/pom.xml
new file mode 100644
index 000000000..03413fc0b
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/pom.xml
@@ -0,0 +1,41 @@
+
+
+
+
+
+ optional-spring-plugins
+ org.apache.skywalking
+ 5.0.0-alpha
+
+ 4.0.0
+ jar
+
+ apm-spring-annotation-plugin
+
+
+
+ org.springframework
+ spring-context
+ 3.2.0.RELEASE
+ provided
+
+
+
\ No newline at end of file
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/AbstractSpringBeanInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/AbstractSpringBeanInstrumentation.java
new file mode 100644
index 000000000..961eab6cb
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/AbstractSpringBeanInstrumentation.java
@@ -0,0 +1,58 @@
+/*
+ * 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.spring.annotations;
+
+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 static net.bytebuddy.matcher.ElementMatchers.isDeclaredBy;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+
+public abstract class AbstractSpringBeanInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+ private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.spring.annotations.SpringAnnotationInterceptor";
+ public static final String INTERCEPT_GET_SKYWALKING_DYNAMIC_FIELD_METHOD = "getSkyWalkingDynamicField";
+ public static final String INTERCEPT_SET_SKYWALKING_DYNAMIC_FEILD_METHOD = "setSkyWalkingDynamicField";
+
+ @Override protected final ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+ return new ConstructorInterceptPoint[0];
+ }
+
+ @Override protected final InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+ return new InstanceMethodsInterceptPoint[] {
+ new InstanceMethodsInterceptPoint() {
+ @Override public ElementMatcher getMethodsMatcher() {
+ return isPublic().and(not(isDeclaredBy(Object.class)).and(not(named(INTERCEPT_GET_SKYWALKING_DYNAMIC_FIELD_METHOD))).and(not(named(INTERCEPT_SET_SKYWALKING_DYNAMIC_FEILD_METHOD))));
+ }
+
+ @Override public String getMethodsInterceptor() {
+ return INTERCEPTOR_CLASS;
+ }
+
+ @Override public boolean isOverrideArgs() {
+ return false;
+ }
+ }
+ };
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/SpringAnnotationInterceptor.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/SpringAnnotationInterceptor.java
new file mode 100644
index 000000000..ad66e1d21
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/SpringAnnotationInterceptor.java
@@ -0,0 +1,46 @@
+/*
+ * 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.spring.annotations;
+
+import java.lang.reflect.Method;
+import org.apache.skywalking.apm.agent.core.context.ContextManager;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
+
+public class SpringAnnotationInterceptor implements InstanceMethodsAroundInterceptor {
+ @Override
+ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes,
+ MethodInterceptResult result) throws Throwable {
+ String operationName = objInst.getClass().getName() + "." + method.getName();
+ ContextManager.createLocalSpan(operationName);
+ }
+
+ @Override
+ public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes,
+ Object ret) throws Throwable {
+ ContextManager.stopSpan();
+ return ret;
+ }
+
+ @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
+ Class[] argumentsTypes, Throwable t) {
+ ContextManager.activeSpan().errorOccurred().log(t);
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/bean/SpringBeanInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/bean/SpringBeanInstrumentation.java
new file mode 100644
index 000000000..8aa5b2277
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/bean/SpringBeanInstrumentation.java
@@ -0,0 +1,36 @@
+/*
+ * 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.spring.annotations.bean;
+
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.spring.annotations.AbstractSpringBeanInstrumentation;
+
+import static org.apache.skywalking.apm.agent.core.plugin.match.ClassAnnotationMatch.byClassAnnotationMatch;
+
+/**
+ *
+ */
+public class SpringBeanInstrumentation extends AbstractSpringBeanInstrumentation {
+
+ public static final String ENHANCE_ANNOTATION = "org.springframework.context.annotation.Bean";
+
+ @Override protected ClassMatch enhanceClass() {
+ return byClassAnnotationMatch(new String[] {ENHANCE_ANNOTATION});
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/component/SpringComponentInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/component/SpringComponentInstrumentation.java
new file mode 100644
index 000000000..eac6bb0c4
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/component/SpringComponentInstrumentation.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spring.annotations.component;
+
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.spring.annotations.AbstractSpringBeanInstrumentation;
+
+import static org.apache.skywalking.apm.agent.core.plugin.match.ClassAnnotationMatch.byClassAnnotationMatch;
+
+public class SpringComponentInstrumentation extends AbstractSpringBeanInstrumentation {
+
+ public static final String ENHANCE_ANNOTATION = "org.springframework.stereotype.Component";
+
+ @Override protected ClassMatch enhanceClass() {
+ return byClassAnnotationMatch(new String[] {ENHANCE_ANNOTATION});
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/repository/SpringRepositoryInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/repository/SpringRepositoryInstrumentation.java
new file mode 100644
index 000000000..d91637e0e
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/repository/SpringRepositoryInstrumentation.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spring.annotations.repository;
+
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.spring.annotations.AbstractSpringBeanInstrumentation;
+
+import static org.apache.skywalking.apm.agent.core.plugin.match.ClassAnnotationMatch.byClassAnnotationMatch;
+
+public class SpringRepositoryInstrumentation extends AbstractSpringBeanInstrumentation {
+
+ public static final String ENHANCE_ANNOTATION = "org.springframework.stereotype.Repository";
+
+ @Override protected ClassMatch enhanceClass() {
+ return byClassAnnotationMatch(new String[] {ENHANCE_ANNOTATION});
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/services/SpringServicesInstrumentation.java b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/services/SpringServicesInstrumentation.java
new file mode 100644
index 000000000..759097ab3
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/annotations/services/SpringServicesInstrumentation.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spring.annotations.services;
+
+import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
+import org.apache.skywalking.apm.plugin.spring.annotations.AbstractSpringBeanInstrumentation;
+
+import static org.apache.skywalking.apm.agent.core.plugin.match.ClassAnnotationMatch.byClassAnnotationMatch;
+
+public class SpringServicesInstrumentation extends AbstractSpringBeanInstrumentation {
+
+ public static final String ENHANCE_ANNOTATION = "org.springframework.stereotype.Service";
+
+ @Override protected ClassMatch enhanceClass() {
+ return byClassAnnotationMatch(new String[] {ENHANCE_ANNOTATION});
+ }
+}
diff --git a/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/resources/skywalking-plugin.def
new file mode 100644
index 000000000..e69e32aa2
--- /dev/null
+++ b/apm-sniffer/optional-plugins/optional-spring-plugins/spring-annotation-plugin/src/main/resources/skywalking-plugin.def
@@ -0,0 +1,4 @@
+spring-annotation=org.apache.skywalking.apm.plugin.spring.annotations.bean.SpringBeanInstrumentation
+spring-annotation=org.apache.skywalking.apm.plugin.spring.annotations.component.SpringComponentInstrumentation
+spring-annotation=org.apache.skywalking.apm.plugin.spring.annotations.repository.SpringRepositoryInstrumentation
+spring-annotation=org.apache.skywalking.apm.plugin.spring.annotations.services.SpringServicesInstrumentation
diff --git a/apm-sniffer/optional-plugins/pom.xml b/apm-sniffer/optional-plugins/pom.xml
new file mode 100644
index 000000000..07061b9d6
--- /dev/null
+++ b/apm-sniffer/optional-plugins/pom.xml
@@ -0,0 +1,114 @@
+
+
+
+
+
+ apm-sniffer
+ org.apache.skywalking
+ 5.0.0-alpha
+
+ 4.0.0
+
+ optional-plugins
+ pom
+
+ ${project.build.directory}/../../../../packages/skywalking-agent/optional-plugins
+
+
+
+ optional-spring-plugins
+
+
+
+
+ org.apache.skywalking
+ apm-agent-core
+ ${project.version}
+ provided
+
+
+ org.apache.skywalking
+ apm-util
+ ${project.version}
+ provided
+
+
+ org.apache.skywalking
+ apm-test-tools
+ ${project.version}
+ test
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-antrun-plugin
+
+
+ package
+
+ run
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ ant-contrib
+ ant-contrib
+ 1.0b3
+
+
+ ant
+ ant
+
+
+
+
+ org.apache.ant
+ ant-nodeps
+ 1.8.1
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/apm-sniffer/pom.xml b/apm-sniffer/pom.xml
index 509de67ce..c5a7607c0 100644
--- a/apm-sniffer/pom.xml
+++ b/apm-sniffer/pom.xml
@@ -36,6 +36,7 @@
apm-sdk-plugin
apm-toolkit-activation
apm-test-tools
+ optional-plugins
diff --git a/docs/README.md b/docs/README.md
index c38355652..8928e771a 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -23,4 +23,7 @@
* Protocol
* [Cross Process Propagation Headers Protocol, v1.0](en/Skywalking-Cross-Process-Propagation-Headers-Protocol-v1.md)
* FAQ
+ * [Why only traces in UI?](cn/FAQ/Why-have-traces-no-others.md)
+ * [Too many GRPC logs in the console](cn/FAQ/Too-many-gRPC-logs.md)
+ * [The trace doesn't continue in kafka consumer side](en/FAQ/kafka-plugin.md)
diff --git a/docs/README_ZH.md b/docs/README_ZH.md
index 8b612efc1..2cf5f1658 100644
--- a/docs/README_ZH.md
+++ b/docs/README_ZH.md
@@ -28,3 +28,4 @@
* FAQ
* [Trace查询有数据，但是没有拓扑图和JVM数据?](cn/FAQ/Why-have-traces-no-others-CN.md)
* [加载探针，Console被GRPC日志刷屏](cn/FAQ/Too-many-gRPC-logs-CN.md)
+ * [Kafka消息消费端链路断裂](cn/FAQ/Kafka-plugin-CN.md)
\ No newline at end of file
diff --git a/docs/Supported-list.md b/docs/Supported-list.md
index ea49a3aa1..b1187e394 100644
--- a/docs/Supported-list.md
+++ b/docs/Supported-list.md
@@ -30,6 +30,7 @@
* [gRPC](https://github.com/grpc/grpc-java) 1.6+
* MQ
* [RocketMQ](https://github.com/apache/rocketmq) 4.x
+ * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 1.0
* NoSQL
* Redis
* [Jedis](https://github.com/xetorthio/jedis) 2.x
@@ -40,11 +41,12 @@
* Service Discovery
* [Netflix Eureka](https://github.com/Netflix/eureka)
* Spring Ecosystem
+ * Spring Bean annotations(@Bean, @Service, @Component, @Repository) 3.x and 4.x (Optional²)
* Spring Core Async SuccessCallback/FailureCallback/ListenableFutureCallback 4.x
* Scheduler
* [Elastic Job](https://github.com/elasticjob/elastic-job) 2.x
* OpenTracing community supported
- * Motan
- * Hprose-java
-¹Required dependencies for these components must be first manually downloaded before being built, due to license incompatibilities. For this reason these components are not by default included in the SkyWalking releases.
\ No newline at end of file
+¹Required dependencies for these components must be first manually downloaded before being built, due to license incompatibilities. For this reason these components are not by default included in the SkyWalking releases.
+
+²These plugins affect the performance or must be used under some conditions, from experiences. So only released in `/optional-plugins`, copy to `/plugins` in order to make them work.
diff --git a/docs/cn/FAQ/Kafka-plugin-CN.md b/docs/cn/FAQ/Kafka-plugin-CN.md
new file mode 100644
index 000000000..02d15a9a6
--- /dev/null
+++ b/docs/cn/FAQ/Kafka-plugin-CN.md
@@ -0,0 +1,7 @@
+**现象** ：
+Kafka消息消费端链路断裂
+
+**原因**：
+Kafka探针只是追踪了对Kafka的拉取动作，而整个后续处理过程不是由kafka consumer发起。故，需要在消费处理的发起点，进行手动埋点。
+
+**解决方法**: 可以通过Application Toolkit中的@Trace标注，或者OpenTracing API进行手动埋点。
diff --git a/docs/cn/FAQ/Why-have-traces-no-others-CN.md b/docs/cn/FAQ/Why-have-traces-no-others-CN.md
index fcab017a3..cf8f2a3d2 100644
--- a/docs/cn/FAQ/Why-have-traces-no-others-CN.md
+++ b/docs/cn/FAQ/Why-have-traces-no-others-CN.md
@@ -2,10 +2,9 @@
- Agent和Collector正常工作，没有异常日志
- 已经对系统进行过访问，Trace查询有数据
- UI除Trace查询页面外，其他页面无数据
-- **页面顶部时间轴和当前系统时间不一致**
原因：
-被监控系统所在的操作系统，未设置为当前时区，导致统计数据汇集时间点偏差。
+Collector和被监控应用的系统主机时间，没有同步。
解决方法：
-设置操作系统时间时区
\ No newline at end of file
+同步各主机操作系统时间。
diff --git a/docs/en/FAQ/Too-many-gRPC-logs.md b/docs/en/FAQ/Too-many-gRPC-logs.md
new file mode 100644
index 000000000..58b4b124c
--- /dev/null
+++ b/docs/en/FAQ/Too-many-gRPC-logs.md
@@ -0,0 +1,6 @@
+**Problem**：
+Too many GRPC log in the console
+
+**Reason**：Skywalking uses the GRPC framework to send data, and the GRPC framework reads log configuration files for log output.
+
+**Resolve**: Add filter to `org.apache.skywalking.apm.dependencies` package in log configuration file
diff --git a/docs/en/FAQ/Why-have-traces-no-others.md b/docs/en/FAQ/Why-have-traces-no-others.md
new file mode 100644
index 000000000..c0db2fe41
--- /dev/null
+++ b/docs/en/FAQ/Why-have-traces-no-others.md
@@ -0,0 +1,9 @@
+**Problem**：
+- There is no abnormal log in Agent log and Collector log,
+- The traces show, but no other info in UI.
+
+**Reason**：
+The operating system where the monitored system is located is not set as the current time zone, causing statistics collection time points to deviate.
+
+**Resolve**：
+Make sure the time is sync in collector servers and monitored application servers.
diff --git a/docs/en/FAQ/kafka-plugin.md b/docs/en/FAQ/kafka-plugin.md
new file mode 100644
index 000000000..7fea78807
--- /dev/null
+++ b/docs/en/FAQ/kafka-plugin.md
@@ -0,0 +1,8 @@
+**Problem**:
+The trace doesn't continue in kafka consumer side.
+
+**Reason**:
+The kafka client is pulling message from server, the plugin also just traces the pull action. As that, you need to do the manual instrument before the pull action, and include the further data process.
+
+**Resolve**:
+Use Application Toolkit libraries to do manual instrumentation. such as `@Trace` annotaion or OpenTracing API.