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

add Kafka Clients 3.x support (#429)

This commit is contained in:
Jiajing LU 2022-12-23 13:48:50 +08:00 committed by GitHub
parent 37a880e903
commit ea1c204c95
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 69 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.md
View File

@ -20,6 +20,7 @@ Release Notes.
* Add tags `mq.message.keys` and `mq.message.tags` for RocketMQ producer span
* Clean the trace context which injected into Pulsar MessageImpl after the instance recycled
* Fix In the higher version of mysql-connector-java 8x, there is an error in the value of db.instance.
* Add support for KafkaClients 3.2.x
#### Documentation

2
apm-sniffer/apm-sdk-plugin/kafka-plugin/pom.xml
View File

@ -28,7 +28,7 @@
<artifactId>apm-kafka-plugin</artifactId>
<properties>
<kafka-clients.version>2.0.1</kafka-clients.version>
<kafka-clients.version>3.2.3</kafka-clients.version>
</properties>
<dependencies>

34
apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka3ConsumerInterceptor.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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.internals.Fetch;
import org.apache.kafka.common.TopicPartition;
import java.util.List;
import java.util.Map;
public class Kafka3ConsumerInterceptor extends KafkaConsumerInterceptor {
@SuppressWarnings({"unchecked"})
@Override
protected Map<TopicPartition, List<ConsumerRecord<?, ?>>> fetchRecords(Object retObj) {
return ((Fetch) retObj).records();
}
}

7
apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/KafkaConsumerInterceptor.java
View File

@ -61,7 +61,7 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto
if (ret == null) {
return ret;
}
Map<TopicPartition, List<ConsumerRecord<?, ?>>> records = (Map<TopicPartition, List<ConsumerRecord<?, ?>>>) ret;
Map<TopicPartition, List<ConsumerRecord<?, ?>>> records = fetchRecords(ret);
//
// The entry span will only be created when the consumer received at least one message.
//
@ -100,6 +100,11 @@ public class KafkaConsumerInterceptor implements InstanceMethodsAroundIntercepto
return ret;
}
@SuppressWarnings({"unchecked"})
protected Map<TopicPartition, List<ConsumerRecord<?, ?>>> fetchRecords(Object retObj) {
return (Map<TopicPartition, List<ConsumerRecord<?, ?>>>) retObj;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {

24
apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java
View File

@ -24,7 +24,10 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterc
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import java.util.Map;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.returns;
import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
@ -45,6 +48,7 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation {
public static final String CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.kafka.ConstructorWithConsumerConfigInterceptPoint";
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 ENHANCE_METHOD = "pollOnce";
public static final String ENHANCE_COMPATIBLE_METHOD = "pollForFetches";
public static final String ENHANCE_CLASS = "org.apache.kafka.clients.consumer.KafkaConsumer";
@ -91,7 +95,8 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD));
// targeting Kafka Client < 3.2
return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class)));
}
@Override
@ -104,6 +109,23 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation {
return false;
}
},
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> 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 boolean isOverrideArgs() {
return false;
}
},
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {

2
docs/en/setup/service-agent/java-agent/Supported-list.md
View File

@ -66,7 +66,7 @@ metrics based on the tracing data.
* [JSONRPC4J](https://github.com/briandilley/jsonrpc4j) 1.2.0 -> 1.6
* MQ
* [RocketMQ](https://github.com/apache/rocketmq) 4.x
* [Kafka](http://kafka.apache.org) 0.11.0.0 -> 2.8.0
* [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3
* [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

3
test/plugin/scenarios/kafka-scenario/support-version.list
View File

@ -25,3 +25,6 @@
2.6.1
2.7.0
2.8.0
3.0.2
3.1.2
3.2.3