From 3feef5ab59b64e0981cbf4268fdd7af81d8f0da6 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Mon, 22 Jul 2024 00:38:50 +0800 Subject: [PATCH 01/23] Fixed an issue where intercept methods for kafka3.7 were invalid --- .../kafka/Kafka3_7ConsumerInterceptor.java | 76 +++++++ .../define/KafkaConsumerInstrumentation.java | 185 ++++++++++-------- 2 files changed, 178 insertions(+), 83 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java new file mode 100644 index 0000000000..3e4149b4ed --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java @@ -0,0 +1,76 @@ +/* + * 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; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.internals.Fetch; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.header.Header; +import org.apache.skywalking.apm.agent.core.context.CarrierItem; +import org.apache.skywalking.apm.agent.core.context.ContextCarrier; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; + +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +public class Kafka3_7ConsumerInterceptor extends KafkaConsumerInterceptor { + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { + if (ret instanceof ConsumerRecords) { + ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); + ConsumerRecords consumerRecords = (ConsumerRecords) ret; + if (consumerRecords.count() == 0) { + return ret; + } + for (ConsumerRecord consumerRecord : consumerRecords) { + if (consumerRecord == null) { + continue; + } + ContextCarrier contextCarrier = new ContextCarrier(); + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + next = next.next(); + Iterator
iterator = consumerRecord.headers().headers(next.getHeadKey()).iterator(); + if (iterator.hasNext()) { + next.setHeadValue(new String(iterator.next().value(), StandardCharsets.UTF_8)); + } + } + String operationName = OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo.getGroupId(); + AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, contextCarrier).start(requiredInfo.getStartTime()); + activeSpan.setComponent(ComponentsDefine.KAFKA_CONSUMER); + SpanLayer.asMQ(activeSpan); + Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); + Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); + activeSpan.setPeer(requiredInfo.getBrokerServers()); + ContextManager.stopSpan(); + } + } + return ret; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index e29996586c..2ac1355cf6 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -50,6 +50,8 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.KafkaConsumerInterceptor"; public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; public static final String ENHANCE_METHOD = "pollOnce"; + public static final String ENHANCE_METHOD_KAFKA3_7 = "poll"; + public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka3_7ConsumerInterceptor"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; public static final String SUBSCRIBE_METHOD = "subscribe"; @@ -62,104 +64,121 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { - return new ConstructorInterceptPoint[] { - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); - } + return new ConstructorInterceptPoint[]{ + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); + } - @Override - public String getConstructorInterceptor() { - return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); - } - - @Override - public String getConstructorInterceptor() { - return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, + @Override + public String getConstructorInterceptor() { + return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); + } + + @Override + public String getConstructorInterceptor() { + return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, }; } @Override public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { - return new InstanceMethodsInterceptPoint[] { - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client < 3.2 - return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); - } + return new InstanceMethodsInterceptPoint[]{ + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client < 3.2 + return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 - return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client >= 3.2 + return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_KAFKA3_2; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_KAFKA3_2; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client >= 3.7 + return named(ENHANCE_METHOD_KAFKA3_7).and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecords"))); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_KAFKA3_7; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); + } + + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { From 2971c2bd02f81cd4ba3ab228c9cf75ba053d3f0f Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Mon, 22 Jul 2024 01:03:30 +0800 Subject: [PATCH 02/23] Fixed an issue where intercept methods for kafka3.7 were invalid --- ...umerInterceptor.java => Kafka37ConsumerInterceptor.java} | 6 +----- .../plugin/kafka/define/KafkaConsumerInstrumentation.java | 2 +- 2 files changed, 2 insertions(+), 6 deletions(-) rename apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/{Kafka3_7ConsumerInterceptor.java => Kafka37ConsumerInterceptor.java} (94%) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java similarity index 94% rename from apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java rename to apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java index 3e4149b4ed..e9f38a5f2b 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java @@ -20,8 +20,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.internals.Fetch; -import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Header; import org.apache.skywalking.apm.agent.core.context.CarrierItem; import org.apache.skywalking.apm.agent.core.context.ContextCarrier; @@ -35,10 +33,8 @@ import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.util.Iterator; -import java.util.List; -import java.util.Map; -public class Kafka3_7ConsumerInterceptor extends KafkaConsumerInterceptor { +public class Kafka37ConsumerInterceptor extends KafkaConsumerInterceptor { @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index 2ac1355cf6..14112575d1 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -51,7 +51,7 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; public static final String ENHANCE_METHOD = "pollOnce"; public static final String ENHANCE_METHOD_KAFKA3_7 = "poll"; - public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka3_7ConsumerInterceptor"; + public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; public static final String SUBSCRIBE_METHOD = "subscribe"; From 7db4f51f0425b5a5d4e1f2b2bd54d462e6bb244b Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Mon, 22 Jul 2024 15:46:35 +0800 Subject: [PATCH 03/23] Revert "Fixed an issue where intercept methods for kafka3.7 were invalid" This reverts commit 2971c2bd02f81cd4ba3ab228c9cf75ba053d3f0f. --- ...merInterceptor.java => Kafka3_7ConsumerInterceptor.java} | 6 +++++- .../plugin/kafka/define/KafkaConsumerInstrumentation.java | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) rename apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/{Kafka37ConsumerInterceptor.java => Kafka3_7ConsumerInterceptor.java} (94%) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java similarity index 94% rename from apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java rename to apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java index e9f38a5f2b..3e4149b4ed 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java @@ -20,6 +20,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.internals.Fetch; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Header; import org.apache.skywalking.apm.agent.core.context.CarrierItem; import org.apache.skywalking.apm.agent.core.context.ContextCarrier; @@ -33,8 +35,10 @@ import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.util.Iterator; +import java.util.List; +import java.util.Map; -public class Kafka37ConsumerInterceptor extends KafkaConsumerInterceptor { +public class Kafka3_7ConsumerInterceptor extends KafkaConsumerInterceptor { @Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index 14112575d1..2ac1355cf6 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -51,7 +51,7 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; public static final String ENHANCE_METHOD = "pollOnce"; public static final String ENHANCE_METHOD_KAFKA3_7 = "poll"; - public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; + public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka3_7ConsumerInterceptor"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; public static final String SUBSCRIBE_METHOD = "subscribe"; From 1cbc3db6e7dbd5e003877b545c876db83123ae97 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Mon, 22 Jul 2024 15:46:47 +0800 Subject: [PATCH 04/23] Revert "Fixed an issue where intercept methods for kafka3.7 were invalid" This reverts commit 3feef5ab59b64e0981cbf4268fdd7af81d8f0da6. --- .../kafka/Kafka3_7ConsumerInterceptor.java | 76 ------- .../define/KafkaConsumerInstrumentation.java | 185 ++++++++---------- 2 files changed, 83 insertions(+), 178 deletions(-) delete mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java deleted file mode 100644 index 3e4149b4ed..0000000000 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3_7ConsumerInterceptor.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * 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; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.internals.Fetch; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.header.Header; -import org.apache.skywalking.apm.agent.core.context.CarrierItem; -import org.apache.skywalking.apm.agent.core.context.ContextCarrier; -import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.tag.Tags; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; -import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; -import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; - -import java.lang.reflect.Method; -import java.nio.charset.StandardCharsets; -import java.util.Iterator; -import java.util.List; -import java.util.Map; - -public class Kafka3_7ConsumerInterceptor extends KafkaConsumerInterceptor { - - @Override - public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { - if (ret instanceof ConsumerRecords) { - ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); - ConsumerRecords consumerRecords = (ConsumerRecords) ret; - if (consumerRecords.count() == 0) { - return ret; - } - for (ConsumerRecord consumerRecord : consumerRecords) { - if (consumerRecord == null) { - continue; - } - ContextCarrier contextCarrier = new ContextCarrier(); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - Iterator
iterator = consumerRecord.headers().headers(next.getHeadKey()).iterator(); - if (iterator.hasNext()) { - next.setHeadValue(new String(iterator.next().value(), StandardCharsets.UTF_8)); - } - } - String operationName = OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo.getGroupId(); - AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, contextCarrier).start(requiredInfo.getStartTime()); - activeSpan.setComponent(ComponentsDefine.KAFKA_CONSUMER); - SpanLayer.asMQ(activeSpan); - Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); - Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); - activeSpan.setPeer(requiredInfo.getBrokerServers()); - ContextManager.stopSpan(); - } - } - return ret; - } -} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index 2ac1355cf6..e29996586c 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -50,8 +50,6 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.KafkaConsumerInterceptor"; public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; public static final String ENHANCE_METHOD = "pollOnce"; - public static final String ENHANCE_METHOD_KAFKA3_7 = "poll"; - public static final String INTERCEPTOR_CLASS_KAFKA3_7 = "org.apache.skywalking.apm.plugin.kafka.Kafka3_7ConsumerInterceptor"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; public static final String SUBSCRIBE_METHOD = "subscribe"; @@ -64,121 +62,104 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { - return new ConstructorInterceptPoint[]{ - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); - } - - @Override - public String getConstructorInterceptor() { - return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); - } + return new ConstructorInterceptPoint[] { + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); + } - @Override - public String getConstructorInterceptor() { - return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, + @Override + public String getConstructorInterceptor() { + return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); + } + + @Override + public String getConstructorInterceptor() { + return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, }; } @Override public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { - return new InstanceMethodsInterceptPoint[]{ - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client < 3.2 - return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); - } - - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS; - } - - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 - return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); - } + return new InstanceMethodsInterceptPoint[] { + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client < 3.2 + return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_KAFKA3_2; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.7 - return named(ENHANCE_METHOD_KAFKA3_7).and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecords"))); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client >= 3.2 + return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_KAFKA3_7; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_KAFKA3_2; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, + @Override + public boolean isOverrideArgs() { + return false; + } + }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { From 6220bd457737299f0a3425567d78f138b104c898 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Mon, 22 Jul 2024 15:53:19 +0800 Subject: [PATCH 05/23] add support for kafka3.7.x, because kafka3.7 changed the pull message method's name from pollForFetches to poll --- .../kafka/Kafka37ConsumerInterceptor.java | 78 +++++++++ .../Kafka37ConsumerInstrumentation.java | 59 +++++++ .../define/KafkaConsumerInstrumentation.java | 162 +++++++++--------- .../src/main/resources/skywalking-plugin.def | 3 +- .../kafka/Kafka37ConsumerInterceptorTest.java | 153 +++++++++++++++++ .../kafka-scenario/support-version.list | 2 + 6 files changed, 375 insertions(+), 82 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java new file mode 100644 index 0000000000..affafb78a8 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java @@ -0,0 +1,78 @@ +/* + * 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; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.header.Header; +import org.apache.skywalking.apm.agent.core.context.CarrierItem; +import org.apache.skywalking.apm.agent.core.context.ContextCarrier; +import org.apache.skywalking.apm.agent.core.context.ContextManager; +import org.apache.skywalking.apm.agent.core.context.tag.Tags; +import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; +import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; +import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; + +import java.lang.reflect.Method; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; + +/** + * kafak3.7 removed the method named pollForFetches with return type Fetch. + * When the kafka version < 3.7, the enhance method could be: org.apache.kafka.clients.consumer.KafkaConsumer#pollForFetches. + * But when the kafka version >= 3.7, the enhance method should be: org.apache.kafka.clients.consumer.KafkaConsumer#poll(java.time.Duration). + * And the return type was also changed, from org.apache.kafka.clients.consumer.internals.Fetch to org.apache.kafka.clients.consumer.ConsumerRecords. + */ +public class Kafka37ConsumerInterceptor extends KafkaConsumerInterceptor { + + @Override + public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { + if (ret instanceof ConsumerRecords) { + ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); + ConsumerRecords consumerRecords = (ConsumerRecords) ret; + if (consumerRecords.count() == 0) { + return ret; + } + for (ConsumerRecord consumerRecord : consumerRecords) { + if (consumerRecord == null) { + continue; + } + ContextCarrier contextCarrier = new ContextCarrier(); + CarrierItem next = contextCarrier.items(); + while (next.hasNext()) { + next = next.next(); + Iterator
iterator = consumerRecord.headers().headers(next.getHeadKey()).iterator(); + if (iterator.hasNext()) { + next.setHeadValue(new String(iterator.next().value(), StandardCharsets.UTF_8)); + } + } + String operationName = OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo.getGroupId(); + AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, contextCarrier).start(requiredInfo.getStartTime()); + activeSpan.setComponent(ComponentsDefine.KAFKA_CONSUMER); + SpanLayer.asMQ(activeSpan); + Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); + Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); + activeSpan.setPeer(requiredInfo.getBrokerServers()); + ContextManager.stopSpan(); + } + } + return ret; + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java new file mode 100644 index 0000000000..765d519c81 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java @@ -0,0 +1,59 @@ +/* + * 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.define; + +import net.bytebuddy.description.method.MethodDescription; +import net.bytebuddy.matcher.ElementMatcher; +import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.returns; + +public class Kafka37ConsumerInstrumentation extends KafkaConsumerInstrumentation { + + public static final String INTERCEPTOR_CLASS_37 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; + + // Kafka 3.7.x's pull message method's name is "poll" + public static final String ENHANCE_METHOD_37 = "poll"; + + // Kafka 3.7.x's pull message method's return type is "ConsumerRecords" + public static final String ENHANCE_RETURN_TYPE_37 = "org.apache.kafka.clients.consumer.ConsumerRecords"; + + @Override + public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { + return new InstanceMethodsInterceptPoint[]{ + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_37; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + } + }; + } +} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index e29996586c..39fa8c1322 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -63,28 +63,28 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { return new ConstructorInterceptPoint[] { - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); - } + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); + } - @Override - public String getConstructorInterceptor() { - return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); - } - - @Override - public String getConstructorInterceptor() { - return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, + @Override + public String getConstructorInterceptor() { + return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); + } + + @Override + public String getConstructorInterceptor() { + return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, }; } @@ -92,74 +92,74 @@ public String getConstructorInterceptor() { @Override public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { return new InstanceMethodsInterceptPoint[] { - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client < 3.2 - return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); - } + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client < 3.2 + return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 - return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client >= 3.2 + return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_KAFKA3_2; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_KAFKA3_2; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, + @Override + public boolean isOverrideArgs() { + return false; + } + }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def index d807aceefe..8c398be2a0 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def @@ -18,4 +18,5 @@ kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.CallbackInstr kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaConsumerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerMapInstrumentation -kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation \ No newline at end of file +kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation +kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37ConsumerInstrumentation \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java new file mode 100644 index 0000000000..fb6cf22096 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java @@ -0,0 +1,153 @@ +/* + * 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; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; +import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem; +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.hamcrest.MatcherAssert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA_CONSUMER; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +@RunWith(TracingSegmentRunner.class) +public class Kafka37ConsumerInterceptorTest { + + @SegmentStoragePoint + private SegmentStorage segmentStorage; + + @Rule + public AgentServiceRule serviceRule = new AgentServiceRule(); + @Rule + public MockitoRule rule = MockitoJUnit.rule(); + + private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo; + + private Kafka37ConsumerInterceptor 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 Kafka37ConsumerInterceptor(); + 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); + consumerEnhanceRequiredInfo.setGroupId("test"); + + 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( + SW8CarrierItem.HEADER_NAME, + "1-My40LjU=-MS4yLjM=-3-c2VydmljZQ==-aW5zdGFuY2U=-L2FwcA==-MTI3LjAuMC4xOjgwODA=" + .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 ConsumerRecords(null)); + + 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], new ConsumerRecords(messages)); + + List traceSegments = segmentStorage.getTraceSegments(); + assertThat(traceSegments.size(), is(1)); + + TraceSegment traceSegment = traceSegments.get(0); + assertNotNull(traceSegment.getRef()); + assertTraceSegmentRef(traceSegment.getRef()); + + 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_CONSUMER); + 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.getParentServiceInstance(ref), is("instance")); + MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3)); + MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("3.4.5")); + } +} diff --git a/test/plugin/scenarios/kafka-scenario/support-version.list b/test/plugin/scenarios/kafka-scenario/support-version.list index 1ffc63cb11..0bb122f2c3 100644 --- a/test/plugin/scenarios/kafka-scenario/support-version.list +++ b/test/plugin/scenarios/kafka-scenario/support-version.list @@ -28,3 +28,5 @@ 3.0.2 3.1.2 3.2.3 +3.7.0 +3.7.1 From 1e2b82a7afde53ffed764a5f8cf9c4fad2d8e243 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 09:41:36 +0800 Subject: [PATCH 06/23] add support for kafka3.7.x, because kafka3.7 changed the pull message method's name from pollForFetches to poll --- .../kafka/Kafka37ConsumerInterceptor.java | 43 +++++++------------ 1 file changed, 15 insertions(+), 28 deletions(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java index affafb78a8..1add769d97 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java @@ -20,6 +20,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Header; import org.apache.skywalking.apm.agent.core.context.CarrierItem; import org.apache.skywalking.apm.agent.core.context.ContextCarrier; @@ -32,7 +33,7 @@ import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; -import java.util.Iterator; +import java.util.*; /** * kafak3.7 removed the method named pollForFetches with return type Fetch. @@ -43,36 +44,22 @@ public class Kafka37ConsumerInterceptor extends KafkaConsumerInterceptor { @Override - public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class[] argumentsTypes, Object ret) throws Throwable { - if (ret instanceof ConsumerRecords) { - ConsumerEnhanceRequiredInfo requiredInfo = (ConsumerEnhanceRequiredInfo) objInst.getSkyWalkingDynamicField(); - ConsumerRecords consumerRecords = (ConsumerRecords) ret; - if (consumerRecords.count() == 0) { - return ret; + protected Map>> fetchRecords(Object retObj) { + Map>> rsp = new HashMap<>(); + if (retObj instanceof ConsumerRecords) { + ConsumerRecords consumerRecords = (ConsumerRecords) retObj; + if (consumerRecords.isEmpty()) { + return rsp; } for (ConsumerRecord consumerRecord : consumerRecords) { - if (consumerRecord == null) { - continue; - } - ContextCarrier contextCarrier = new ContextCarrier(); - CarrierItem next = contextCarrier.items(); - while (next.hasNext()) { - next = next.next(); - Iterator
iterator = consumerRecord.headers().headers(next.getHeadKey()).iterator(); - if (iterator.hasNext()) { - next.setHeadValue(new String(iterator.next().value(), StandardCharsets.UTF_8)); - } - } - String operationName = OPERATE_NAME_PREFIX + requiredInfo.getTopics() + CONSUMER_OPERATE_NAME + requiredInfo.getGroupId(); - AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, contextCarrier).start(requiredInfo.getStartTime()); - activeSpan.setComponent(ComponentsDefine.KAFKA_CONSUMER); - SpanLayer.asMQ(activeSpan); - Tags.MQ_BROKER.set(activeSpan, requiredInfo.getBrokerServers()); - Tags.MQ_TOPIC.set(activeSpan, requiredInfo.getTopics()); - activeSpan.setPeer(requiredInfo.getBrokerServers()); - ContextManager.stopSpan(); + int partition = consumerRecord.partition(); + String topic = consumerRecord.topic(); + TopicPartition topicPartition = new TopicPartition(topic, partition); + List> recordsForTp = rsp.computeIfAbsent(topicPartition, k -> new ArrayList<>()); + recordsForTp.add(consumerRecord); } } - return ret; + return rsp; } + } \ No newline at end of file From d875bd38b58d2b3f47bb05acf85a1f5d305347ce Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 09:44:32 +0800 Subject: [PATCH 07/23] add support for kafka3.7.x, because kafka3.7 changed the pull message method's name from pollForFetches to poll --- .../plugin/kafka/Kafka37ConsumerInterceptor.java | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java index 1add769d97..c9a082ec54 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java @@ -21,19 +21,11 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.header.Header; -import org.apache.skywalking.apm.agent.core.context.CarrierItem; -import org.apache.skywalking.apm.agent.core.context.ContextCarrier; -import org.apache.skywalking.apm.agent.core.context.ContextManager; -import org.apache.skywalking.apm.agent.core.context.tag.Tags; -import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan; -import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance; -import org.apache.skywalking.apm.network.trace.component.ComponentsDefine; -import java.lang.reflect.Method; -import java.nio.charset.StandardCharsets; -import java.util.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; /** * kafak3.7 removed the method named pollForFetches with return type Fetch. From ef2193b34bc6ce9fd38a5c327650eb4d0a45bdad Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 10:02:43 +0800 Subject: [PATCH 08/23] add support for kafka3.7.x, because kafka3.7 changed the pull message method's name from pollForFetches to poll --- .../Kafka37ConsumerInstrumentation.java | 59 ------------------- .../define/KafkaConsumerInstrumentation.java | 27 ++++++++- .../src/main/resources/skywalking-plugin.def | 1 - 3 files changed, 26 insertions(+), 61 deletions(-) delete mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java deleted file mode 100644 index 765d519c81..0000000000 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * 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.define; - -import net.bytebuddy.description.method.MethodDescription; -import net.bytebuddy.matcher.ElementMatcher; -import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint; - -import static net.bytebuddy.matcher.ElementMatchers.named; -import static net.bytebuddy.matcher.ElementMatchers.returns; - -public class Kafka37ConsumerInstrumentation extends KafkaConsumerInstrumentation { - - public static final String INTERCEPTOR_CLASS_37 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; - - // Kafka 3.7.x's pull message method's name is "poll" - public static final String ENHANCE_METHOD_37 = "poll"; - - // Kafka 3.7.x's pull message method's return type is "ConsumerRecords" - public static final String ENHANCE_RETURN_TYPE_37 = "org.apache.kafka.clients.consumer.ConsumerRecords"; - - @Override - public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { - return new InstanceMethodsInterceptPoint[]{ - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))); - } - - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_37; - } - - @Override - public boolean isOverrideArgs() { - return false; - } - } - }; - } -} diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index 39fa8c1322..aed5f9a06c 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -49,6 +49,15 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String MAP_CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.ConstructorWithMapInterceptPoint"; public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.KafkaConsumerInterceptor"; public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; + + public static final String INTERCEPTOR_CLASS_37 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; + + // Kafka 3.7.x's pull message method's name is "poll" + public static final String ENHANCE_METHOD_37 = "poll"; + + // Kafka 3.7.x's pull message method's return type is "ConsumerRecords" + public static final String ENHANCE_RETURN_TYPE_37 = "org.apache.kafka.clients.consumer.ConsumerRecords"; + public static final String ENHANCE_METHOD = "pollOnce"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; @@ -112,7 +121,7 @@ public boolean isOverrideArgs() { new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 + // targeting Kafka Client >= 3.2 and < 3.7.x return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); } @@ -126,6 +135,22 @@ public boolean isOverrideArgs() { return false; } }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))); + } + + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_37; + } + + @Override + public boolean isOverrideArgs() { + return false; + } + }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def index 8c398be2a0..a7e395903a 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def @@ -19,4 +19,3 @@ kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaConsumer kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerMapInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation -kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37ConsumerInstrumentation \ No newline at end of file From e20ef76b1cc00c1db1a8920539674925f734ad6d Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 11:37:42 +0800 Subject: [PATCH 09/23] repair ci error --- .../plugin/kafka/define/KafkaConsumerInstrumentation.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index aed5f9a06c..a50cb560b3 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -55,6 +55,8 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { // Kafka 3.7.x's pull message method's name is "poll" public static final String ENHANCE_METHOD_37 = "poll"; + public static final String ENHANCE_PARAM_TYPE_0_37 = "java.time.Duration"; + // Kafka 3.7.x's pull message method's return type is "ConsumerRecords" public static final String ENHANCE_RETURN_TYPE_37 = "org.apache.kafka.clients.consumer.ConsumerRecords"; @@ -121,7 +123,7 @@ public boolean isOverrideArgs() { new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 and < 3.7.x + // targeting Kafka Client >= 3.2 and < 3.7.0 return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); } @@ -136,9 +138,11 @@ public boolean isOverrideArgs() { } }, new InstanceMethodsInterceptPoint() { + // targeting Kafka Client >= 3.7.0 @Override public ElementMatcher getMethodsMatcher() { - return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))); + return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))) + .and(takesArgumentWithType(0, ENHANCE_PARAM_TYPE_0_37)); } @Override From 354320447575451d305bc91d1e628d47b23d38a8 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 11:46:40 +0800 Subject: [PATCH 10/23] add case of 3.6.0 --- test/plugin/scenarios/kafka-scenario/support-version.list | 1 + 1 file changed, 1 insertion(+) diff --git a/test/plugin/scenarios/kafka-scenario/support-version.list b/test/plugin/scenarios/kafka-scenario/support-version.list index 0bb122f2c3..6e83b29a43 100644 --- a/test/plugin/scenarios/kafka-scenario/support-version.list +++ b/test/plugin/scenarios/kafka-scenario/support-version.list @@ -28,5 +28,6 @@ 3.0.2 3.1.2 3.2.3 +3.6.0 3.7.0 3.7.1 From 526abbb257ea9bed231ab0edba2bdf46ed71b031 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Tue, 23 Jul 2024 12:02:39 +0800 Subject: [PATCH 11/23] fix upe --- .../apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java index fb6cf22096..de26dfa86b 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java @@ -114,7 +114,7 @@ public void setUp() { 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 ConsumerRecords(null)); + consumerInstance, null, new Object[0], new Class[0], new ConsumerRecords(new HashMap<>())); List traceSegments = segmentStorage.getTraceSegments(); assertThat(traceSegments.size(), is(0)); From abd38489634c814e606e9b66f5ab18218ac3b6d1 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 10:20:50 +0800 Subject: [PATCH 12/23] repair spring-kafka 2.2.x ci --- .../Kafka37AsyncConsumerInstrumentation.java | 42 +++++++++++++++++++ .../Kafka37LegacyConsumerInstrumentation.java | 42 +++++++++++++++++++ .../define/KafkaConsumerInstrumentation.java | 33 +-------------- .../src/main/resources/skywalking-plugin.def | 2 + 4 files changed, 88 insertions(+), 31 deletions(-) create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java create mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java new file mode 100644 index 0000000000..566f925abf --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java @@ -0,0 +1,42 @@ +/* + * 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.define; + +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * Here is the intercept process steps. + * + * 
+ *  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.
+ *
+ */ +public class Kafka37AsyncConsumerInstrumentation extends KafkaConsumerInstrumentation { + + private static final String ENHANCE_CLASS_37_ASYNC = "org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS_37_ASYNC); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java new file mode 100644 index 0000000000..4db4f42b68 --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java @@ -0,0 +1,42 @@ +/* + * 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.define; + +import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch; +import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; + +/** + * Here is the intercept process steps. + * + * 
+ *  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.
+ *
+ */ +public class Kafka37LegacyConsumerInstrumentation extends KafkaConsumerInstrumentation { + + private static final String ENHANCE_CLASS_37_LEGACY = "org.apache.kafka.clients.consumer.internals.LegacyKafkaConsumer"; + + @Override + protected ClassMatch enhanceClass() { + return byName(ENHANCE_CLASS_37_LEGACY); + } +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index a50cb560b3..9a9877dcc8 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -49,17 +49,6 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { public static final String MAP_CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.ConstructorWithMapInterceptPoint"; public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.KafkaConsumerInterceptor"; public static final String INTERCEPTOR_CLASS_KAFKA3_2 = "org.apache.skywalking.apm.plugin.kafka.Kafka3ConsumerInterceptor"; - - public static final String INTERCEPTOR_CLASS_37 = "org.apache.skywalking.apm.plugin.kafka.Kafka37ConsumerInterceptor"; - - // Kafka 3.7.x's pull message method's name is "poll" - public static final String ENHANCE_METHOD_37 = "poll"; - - public static final String ENHANCE_PARAM_TYPE_0_37 = "java.time.Duration"; - - // Kafka 3.7.x's pull message method's return type is "ConsumerRecords" - public static final String ENHANCE_RETURN_TYPE_37 = "org.apache.kafka.clients.consumer.ConsumerRecords"; - public static final String ENHANCE_METHOD = "pollOnce"; public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches"; public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer"; @@ -123,7 +112,7 @@ public boolean isOverrideArgs() { new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 and < 3.7.0 + // targeting Kafka Client >= 3.2 return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); } @@ -137,24 +126,6 @@ public boolean isOverrideArgs() { return false; } }, - new InstanceMethodsInterceptPoint() { - // targeting Kafka Client >= 3.7.0 - @Override - public ElementMatcher getMethodsMatcher() { - return named(ENHANCE_METHOD_37).and(returns(named(ENHANCE_RETURN_TYPE_37))) - .and(takesArgumentWithType(0, ENHANCE_PARAM_TYPE_0_37)); - } - - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_37; - } - - @Override - public boolean isOverrideArgs() { - return false; - } - }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { @@ -213,4 +184,4 @@ public boolean isOverrideArgs() { protected ClassMatch enhanceClass() { return byName(ENHANCE_CLASS); } -} +} \ No newline at end of file diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def index a7e395903a..a85deb260b 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def @@ -19,3 +19,5 @@ kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaConsumer kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaProducerMapInstrumentation kafka-0.11.x/1.x/2.x=org.apache.skywalking.apm.plugin.kafka.define.KafkaTemplateCallbackInstrumentation +kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37AsyncConsumerInstrumentation +kafka-3.7.x=org.apache.skywalking.apm.plugin.kafka.define.Kafka37LegacyConsumerInstrumentation From 0cb524cf9e57ad3b304f321ac8a750a4df574673 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 10:46:35 +0800 Subject: [PATCH 13/23] remove dead line in Bootstrap-plugins.md --- docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md | 1 - 1 file changed, 1 deletion(-) diff --git a/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md b/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md index 10dd955a9c..32b398aa6f 100644 --- a/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md +++ b/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md @@ -19,4 +19,3 @@ The plugin of JDK HttpURLConnection depended on `sun.net.*`. When using Java 9+, For more information 1. [JEP 403: Strongly Encapsulate JDK Internals](https://openjdk.org/jeps/403) -2. [A peek into Java 17: Encapsulating the Java runtime internals](https://blogs.oracle.com/javamagazine/post/a-peek-into-java-17-continuing-the-drive-to-encapsulate-the-java-runtime-internals) \ No newline at end of file From f31fb6fff7a87d5878302431c85490a33b8c95cb Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 10:48:59 +0800 Subject: [PATCH 14/23] repair the uncompatible code in kafka Case --- .../testcase/kafka/controller/CaseController.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java b/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java index ea8bb6e3c1..ee6458bdc1 100644 --- a/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java +++ b/test/plugin/scenarios/kafka-scenario/src/main/java/test/apache/skywalking/apm/testcase/kafka/controller/CaseController.java @@ -24,6 +24,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import java.util.Collection; import java.util.regex.Pattern; import java.util.List; import java.util.ArrayList; @@ -32,10 +33,10 @@ import okhttp3.OkHttpClient; import okhttp3.Request; import okhttp3.Response; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; @@ -270,7 +271,17 @@ public void run() { consumerProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumerProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer consumer = new KafkaConsumer<>(consumerProperties); - consumer.subscribe(topicPattern, new NoOpConsumerRebalanceListener()); + consumer.subscribe(topicPattern, new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection collection) { + + } + + @Override + public void onPartitionsAssigned(Collection collection) { + + } + }); while (true) { if (pollAndInvoke(consumer)) break; } From 8297b83e487d45083aeb4bc1ddd4e8224498324c Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 11:11:48 +0800 Subject: [PATCH 15/23] test ci for 3.7 jdk ci --- docs/en/setup/service-agent/java-agent/Plugin-list.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/en/setup/service-agent/java-agent/Plugin-list.md b/docs/en/setup/service-agent/java-agent/Plugin-list.md index e15b3b1f00..d2c07c5a41 100644 --- a/docs/en/setup/service-agent/java-agent/Plugin-list.md +++ b/docs/en/setup/service-agent/java-agent/Plugin-list.md @@ -54,6 +54,7 @@ - jetty-client-9.x - jetty-server-9.x - kafka-0.11.x/1.x/2.x +- kafka-3.7.x - kotlin-coroutine - lettuce-5.x - light4j From 73c85a01edb67650259aabe8aba24e780665929c Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 11:36:58 +0800 Subject: [PATCH 16/23] add support version in Supported-list.md --- .../kafka/Kafka37ConsumerInterceptor.java | 57 ------------------- .../java-agent/Bootstrap-plugins.md | 1 + .../java-agent/Supported-list.md | 2 +- 3 files changed, 2 insertions(+), 58 deletions(-) delete mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java deleted file mode 100644 index c9a082ec54..0000000000 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * 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; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.TopicPartition; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -/** - * kafak3.7 removed the method named pollForFetches with return type Fetch. - * When the kafka version < 3.7, the enhance method could be: org.apache.kafka.clients.consumer.KafkaConsumer#pollForFetches. - * But when the kafka version >= 3.7, the enhance method should be: org.apache.kafka.clients.consumer.KafkaConsumer#poll(java.time.Duration). - * And the return type was also changed, from org.apache.kafka.clients.consumer.internals.Fetch to org.apache.kafka.clients.consumer.ConsumerRecords. - */ -public class Kafka37ConsumerInterceptor extends KafkaConsumerInterceptor { - - @Override - protected Map>> fetchRecords(Object retObj) { - Map>> rsp = new HashMap<>(); - if (retObj instanceof ConsumerRecords) { - ConsumerRecords consumerRecords = (ConsumerRecords) retObj; - if (consumerRecords.isEmpty()) { - return rsp; - } - for (ConsumerRecord consumerRecord : consumerRecords) { - int partition = consumerRecord.partition(); - String topic = consumerRecord.topic(); - TopicPartition topicPartition = new TopicPartition(topic, partition); - List> recordsForTp = rsp.computeIfAbsent(topicPartition, k -> new ArrayList<>()); - recordsForTp.add(consumerRecord); - } - } - return rsp; - } - -} \ No newline at end of file diff --git a/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md b/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md index 32b398aa6f..10dd955a9c 100644 --- a/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md +++ b/docs/en/setup/service-agent/java-agent/Bootstrap-plugins.md @@ -19,3 +19,4 @@ The plugin of JDK HttpURLConnection depended on `sun.net.*`. When using Java 9+, For more information 1. [JEP 403: Strongly Encapsulate JDK Internals](https://openjdk.org/jeps/403) +2. [A peek into Java 17: Encapsulating the Java runtime internals](https://blogs.oracle.com/javamagazine/post/a-peek-into-java-17-continuing-the-drive-to-encapsulate-the-java-runtime-internals) \ No newline at end of file diff --git a/docs/en/setup/service-agent/java-agent/Supported-list.md b/docs/en/setup/service-agent/java-agent/Supported-list.md index c91c25d4a2..235380dd22 100644 --- a/docs/en/setup/service-agent/java-agent/Supported-list.md +++ b/docs/en/setup/service-agent/java-agent/Supported-list.md @@ -76,7 +76,7 @@ metrics based on the tracing data. * MQ * [RocketMQ](https://github.com/apache/rocketmq) 3.x-> 5.x * [RocketMQ-gRPC](http://github.com/apache/rocketmq-clients) 5.x - * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3 + * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3 ->3.7.1 * [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka)) * [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4 * [RabbitMQ](https://www.rabbitmq.com/) 3.x-> 5.x From 1cc94388884568cbb1d353d9c592bbe18fa8010d Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 11:46:35 +0800 Subject: [PATCH 17/23] remove error class --- .../kafka/Kafka37ConsumerInterceptorTest.java | 153 ------------------ 1 file changed, 153 deletions(-) delete mode 100644 apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java deleted file mode 100644 index de26dfa86b..0000000000 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/test/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptorTest.java +++ /dev/null @@ -1,153 +0,0 @@ -/* - * 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; - -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.TopicPartition; -import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem; -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.hamcrest.MatcherAssert; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.junit.MockitoJUnit; -import org.mockito.junit.MockitoRule; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import static org.apache.skywalking.apm.network.trace.component.ComponentsDefine.KAFKA_CONSUMER; -import static org.hamcrest.CoreMatchers.is; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertThat; - -@RunWith(TracingSegmentRunner.class) -public class Kafka37ConsumerInterceptorTest { - - @SegmentStoragePoint - private SegmentStorage segmentStorage; - - @Rule - public AgentServiceRule serviceRule = new AgentServiceRule(); - @Rule - public MockitoRule rule = MockitoJUnit.rule(); - - private ConsumerEnhanceRequiredInfo consumerEnhanceRequiredInfo; - - private Kafka37ConsumerInterceptor 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 Kafka37ConsumerInterceptor(); - 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); - consumerEnhanceRequiredInfo.setGroupId("test"); - - 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( - SW8CarrierItem.HEADER_NAME, - "1-My40LjU=-MS4yLjM=-3-c2VydmljZQ==-aW5zdGFuY2U=-L2FwcA==-MTI3LjAuMC4xOjgwODA=" - .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 ConsumerRecords(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], new ConsumerRecords(messages)); - - List traceSegments = segmentStorage.getTraceSegments(); - assertThat(traceSegments.size(), is(1)); - - TraceSegment traceSegment = traceSegments.get(0); - assertNotNull(traceSegment.getRef()); - assertTraceSegmentRef(traceSegment.getRef()); - - 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_CONSUMER); - 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.getParentServiceInstance(ref), is("instance")); - MatcherAssert.assertThat(SegmentRefHelper.getSpanId(ref), is(3)); - MatcherAssert.assertThat(SegmentRefHelper.getTraceSegmentId(ref).toString(), is("3.4.5")); - } -} From 960be6cb2bc672c3b48fae06c0d2725be5af11ba Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 12:18:03 +0800 Subject: [PATCH 18/23] remove error class --- docs/en/setup/service-agent/java-agent/Supported-list.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/en/setup/service-agent/java-agent/Supported-list.md b/docs/en/setup/service-agent/java-agent/Supported-list.md index 235380dd22..6aa5bd0aa7 100644 --- a/docs/en/setup/service-agent/java-agent/Supported-list.md +++ b/docs/en/setup/service-agent/java-agent/Supported-list.md @@ -76,7 +76,7 @@ metrics based on the tracing data. * MQ * [RocketMQ](https://github.com/apache/rocketmq) 3.x-> 5.x * [RocketMQ-gRPC](http://github.com/apache/rocketmq-clients) 5.x - * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3 ->3.7.1 + * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3 -> 3.7.1 * [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka)) * [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4 * [RabbitMQ](https://www.rabbitmq.com/) 3.x-> 5.x From 14ebf3714d8a5afe3a5a92e83d48db7eb7b167b0 Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 13:59:31 +0800 Subject: [PATCH 19/23] Add comments for resolve the problem that Kafka3.7.x can not be intercepted --- .../kafka/define/Kafka37AsyncConsumerInstrumentation.java | 8 +++----- .../define/Kafka37LegacyConsumerInstrumentation.java | 8 +++----- 2 files changed, 6 insertions(+), 10 deletions(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java index 566f925abf..7c24a135ac 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37AsyncConsumerInstrumentation.java @@ -22,13 +22,11 @@ import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; /** - * Here is the intercept process steps. + * For Kafka 3.7.x change * * 
- *  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.
+ *  1. The method named pollForFetchs was removed from KafkaConsumer to AsyncKafkaConsumer and LegacyKafkaConsumer
+ *  2. Because of the enhance class was changed, so we should create new Instrumentation to intercept the method
  *
*/ public class Kafka37AsyncConsumerInstrumentation extends KafkaConsumerInstrumentation { diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java index 4db4f42b68..95f33bfe82 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37LegacyConsumerInstrumentation.java @@ -22,13 +22,11 @@ import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName; /** - * Here is the intercept process steps. + * For Kafka 3.7.x change * * 
- *  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.
+ *  1. The method named pollForFetchs was removed from KafkaConsumer to AsyncKafkaConsumer and LegacyKafkaConsumer
+ *  2. Because of the enhance class was changed, so we should create new Instrumentation to intercept the method
  *
*/ public class Kafka37LegacyConsumerInstrumentation extends KafkaConsumerInstrumentation { From 39353e848d09333bd22aaf7fd5503fa8e0c1f7ba Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 14:18:10 +0800 Subject: [PATCH 20/23] revert the format codes --- .../define/KafkaConsumerInstrumentation.java | 164 +++++++++--------- 1 file changed, 82 insertions(+), 82 deletions(-) diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java index 9a9877dcc8..e29996586c 100644 --- a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java @@ -63,28 +63,28 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation { @Override public ConstructorInterceptPoint[] getConstructorsInterceptPoints() { return new ConstructorInterceptPoint[] { - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); - } - - @Override - public String getConstructorInterceptor() { - return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, - new ConstructorInterceptPoint() { - @Override - public ElementMatcher getConstructorMatcher() { - return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); - } + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE); + } - @Override - public String getConstructorInterceptor() { - return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; - } - }, + @Override + public String getConstructorInterceptor() { + return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, + new ConstructorInterceptPoint() { + @Override + public ElementMatcher getConstructorMatcher() { + return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE); + } + + @Override + public String getConstructorInterceptor() { + return MAP_CONSTRUCTOR_INTERCEPTOR_CLASS; + } + }, }; } @@ -92,74 +92,74 @@ public String getConstructorInterceptor() { @Override public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() { return new InstanceMethodsInterceptPoint[] { - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client < 3.2 - return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); - } + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client < 3.2 + return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - // targeting Kafka Client >= 3.2 - return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + // targeting Kafka Client >= 3.2 + return named(ENHANCE_COMPATIBLE_METHOD).and(returns(named("org.apache.kafka.clients.consumer.internals.Fetch"))); + } - @Override - public String getMethodsInterceptor() { - return INTERCEPTOR_CLASS_KAFKA3_2; - } + @Override + public String getMethodsInterceptor() { + return INTERCEPTOR_CLASS_KAFKA3_2; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, - new InstanceMethodsInterceptPoint() { - @Override - public ElementMatcher getMethodsMatcher() { - return named(SUBSCRIBE_METHOD) - .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); - } + @Override + public boolean isOverrideArgs() { + return false; + } + }, + new InstanceMethodsInterceptPoint() { + @Override + public ElementMatcher getMethodsMatcher() { + return named(SUBSCRIBE_METHOD) + .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN)); + } - @Override - public String getMethodsInterceptor() { - return SUBSCRIBE_INTERCEPT_CLASS; - } + @Override + public String getMethodsInterceptor() { + return SUBSCRIBE_INTERCEPT_CLASS; + } - @Override - public boolean isOverrideArgs() { - return false; - } - }, + @Override + public boolean isOverrideArgs() { + return false; + } + }, new InstanceMethodsInterceptPoint() { @Override public ElementMatcher getMethodsMatcher() { @@ -184,4 +184,4 @@ public boolean isOverrideArgs() { protected ClassMatch enhanceClass() { return byName(ENHANCE_CLASS); } -} \ No newline at end of file +} From 3f51753400046725539dbcbd29656d1c9cab9079 Mon Sep 17 00:00:00 2001 From: kezhenxu94 Date: Thu, 25 Jul 2024 14:20:34 +0800 Subject: [PATCH 21/23] ignore 403 when checking markdown link --- .dlc.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.dlc.json b/.dlc.json index 1ae0883a40..6a207d2c58 100644 --- a/.dlc.json +++ b/.dlc.json @@ -36,6 +36,7 @@ 200, 301, 302, - 401 + 401, + 403 ] } From 1a5f39cfebbd79af12b293259d9c4c630e96f7cd Mon Sep 17 00:00:00 2001 From: darkness <422019747@qq.com> Date: Thu, 25 Jul 2024 15:19:04 +0800 Subject: [PATCH 22/23] Update CHANGES.md: Support kafka-clients-3.7.x intercept --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 623b5015f1..d2ab11f4ee 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -22,7 +22,7 @@ Release Notes. * Improve 4x performance of ContextManagerExtendService.createTraceContext() * Add a plugin that supports the Solon framework. * Fixed issues in the MySQL component where the executeBatch method could result in empty SQL statements . - +* Support kafka-clients-3.7.x intercept All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/213?closed=1) From b8767658a8db6e4f071fc388170ed800d0286029 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Thu, 25 Jul 2024 15:24:10 +0800 Subject: [PATCH 23/23] Update docs/en/setup/service-agent/java-agent/Supported-list.md --- docs/en/setup/service-agent/java-agent/Supported-list.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/en/setup/service-agent/java-agent/Supported-list.md b/docs/en/setup/service-agent/java-agent/Supported-list.md index 6aa5bd0aa7..cd7c02f78b 100644 --- a/docs/en/setup/service-agent/java-agent/Supported-list.md +++ b/docs/en/setup/service-agent/java-agent/Supported-list.md @@ -76,7 +76,7 @@ metrics based on the tracing data. * MQ * [RocketMQ](https://github.com/apache/rocketmq) 3.x-> 5.x * [RocketMQ-gRPC](http://github.com/apache/rocketmq-clients) 5.x - * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3 -> 3.7.1 + * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.7.1 * [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka)) * [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4 * [RabbitMQ](https://www.rabbitmq.com/) 3.x-> 5.x