diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md deleted file mode 100644 index bcc6c3b46..000000000 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/README.md +++ /dev/null @@ -1,5 +0,0 @@ -# Kafka Plugin - -## -## Buired Point - diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java index f0098e43d..b7bf23b53 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaConsumerInterceptor.java @@ -56,7 +56,7 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto Object ret) throws Throwable { Map>> records = (Map>>)ret; // - // The entry span will create when the consumer fetch anyone message from kafka cluster, or the span will not create. + // The entry span will only be created when the consumer received at least one message. // if (records.size() > 0) { ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo)objInst.getSkyWalkingDynamicField(); diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java index 19d883b70..2e32d4f08 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/KafkaProducerInterceptor.java @@ -50,13 +50,11 @@ public class KafkaProducerInterceptor implements InstanceMethodsAroundIntercepto AbstractSpan activeSpan = ContextManager.createExitSpan(OPERATE_NAME_PREFIX + topicName + PRODUCER_OPERATE_NAME_SUFFIX, contextCarrier, (String)objInst.getSkyWalkingDynamicField()); - //set tags Tags.MQ_BROKER.set(activeSpan, (String)objInst.getSkyWalkingDynamicField()); Tags.MQ_TOPIC.set(activeSpan, topicName); SpanLayer.asMQ(activeSpan); activeSpan.setComponent(ComponentsDefine.KAFKA); - // set headers CarrierItem next = contextCarrier.items(); while (next.hasNext()) { next = next.next(); diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java index 9849d48ff..9dbeba457 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerConstructorInterceptor.java @@ -27,7 +27,6 @@ public class ProducerConstructorInterceptor implements InstanceConstructorInterc @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { ProducerConfig config = (ProducerConfig)allArguments[0]; - // set the bootstrap server address objInst.setSkyWalkingDynamicField(StringUtil.join(';', config.getList("bootstrap.servers").toArray(new String[0]))); } } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java index 0cb723a43..8a7fbac87 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/ProducerRecordConstructorInterceptor.java @@ -24,7 +24,6 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceC public class ProducerRecordConstructorInterceptor implements InstanceConstructorInterceptor { @Override public void onConstruct(EnhancedInstance objInst, Object[] allArguments) { String topic = (String)allArguments[0]; - // set the topic objInst.setSkyWalkingDynamicField(topic); } } diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java index 715166888..f3f22e44a 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/CallbackInstrumentation.java @@ -29,7 +29,8 @@ import static net.bytebuddy.matcher.ElementMatchers.named; import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch; /** - * {@link CallbackInstrumentation} intercept the method onCompletion in the class org.apache.kafka.clients.producer.Callback. + * {@link CallbackInstrumentation} defined that {@link org.apache.skywalking.apm.plugin.kafka.v11.CallbackInterceptor} + * intercept the method onCompletion in the class org.apache.kafka.clients.producer.Callback. * * @author zhangxin */ diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java index 81b9bbe87..f14befb7f 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaConsumerInstrumentation.java @@ -30,34 +30,35 @@ import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentType import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; /** - * {@link KafkaProducerInstrumentation} intercept the method send in the class - * org.apache.kafka.clients.producer.KafkaProducer. Here is the intercept process steps. + * {@link KafkaProducerInstrumentation} define that {@link org.apache.skywalking.apm.plugin.kafka.v11.KafkaConsumerInterceptor} + * intercept the method send in the class org.apache.kafka.clients.producer.KafkaProducer. + * Here is the intercept process steps. * * *
- *  1. Record the topic when the client call subscribed
- *  3. Create the entry span when the client call the method pollOnce.
- *  4. Inject all the Trace Context by iterate all ConsumerRecord
- *  5. Stop the entry span when end the pollOnce method.
+ *  1. Record the topic when the client invoke subscribed method
+ *  2. Create the entry span when the client invoke the method pollOnce.
+ *  3. Extract all the Trace Context by iterate all ConsumerRecord
+ *  4. Stop the entry span when pollOnce method finished.
  * 
* * @author zhang xin */ public class KafkaConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine { - public static final String CONSTRUCTOR_INTERCEPT_FLAG = "org.apache.kafka.clients.consumer.ConsumerConfig"; + public static final String CONSTRUCTOR_INTERCEPT_TYPE = "org.apache.kafka.clients.consumer.ConsumerConfig"; public static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.ConsumerConstructorInterceptor"; public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.v11.KafkaConsumerInterceptor"; public static final String ENHANCE_METHOD = "pollOnce"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; public static final String SUBSCRIBE_METHOD = "subscribe"; - public static final String SUBSCRIBE_INTERCEPT_FLAG = "org.apache.kafka.clients.consumer.ConsumerRebalanceListener"; + public static final String SUBSCRIBE_INTERCEPT_TYPE = "org.apache.kafka.clients.consumer.ConsumerRebalanceListener"; @Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() { return new ConstructorInterceptPoint[] { new ConstructorInterceptPoint() { @Override public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_FLAG); + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); } @Override public String getConstructorInterceptor() { @@ -84,7 +85,7 @@ public class KafkaConsumerInstrumentation extends ClassInstanceMethodsEnhancePlu }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD).and(takesArgumentWithType(1, SUBSCRIBE_INTERCEPT_FLAG)); + return named(SUBSCRIBE_METHOD).and(takesArgumentWithType(1, SUBSCRIBE_INTERCEPT_TYPE)); } @Override public String getMethodsInterceptor() { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java index cd3ff6041..1cca62467 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/KafkaProducerInstrumentation.java @@ -30,18 +30,17 @@ import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentType import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; /** - * {@link KafkaProducerInstrumentation} intercept the method send in the class - * org.apache.kafka.clients.producer.KafkaProducer. Here is the intercept process steps. + * {@link KafkaProducerInstrumentation} define that {@link org.apache.skywalking.apm.plugin.kafka.v11.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. Fetch the topic name from org.apache.kafka.clients.producer.ProducerRecord when the client call
- * send method.
- *  3. Create the exit span when the client call send method
- *  4. Set the Context into the org.apache.kafka.clients.producer.ProducerRecord#headers
- *  5. Stop the exit span when end the send method.
+ *  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 diff --git a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java index abcee1a0a..cb0a052b9 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-0.11.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/v11/define/ProducerRecordInstrumentation.java @@ -29,8 +29,9 @@ import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; /** - * {@link ProducerRecordInstrumentation} intercept the constructor in the class org.apache.kafka.clients.producer.ProducerRecord - * for record the topic name and propagate the Context of trace. + * {@link ProducerRecordInstrumentation} define that {@link org.apache.skywalking.apm.plugin.kafka.v11.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.v11.define.KafkaProducerInstrumentation