add support for kafka3.7.x, because kafka3.7 changed the pull message…

… method's name from pollForFetches to poll
apache · Jul 23, 2024 · ef2193b · ef2193b
1 parent d875bd3
commit ef2193b
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 61 deletions.
diff --git a/...in/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java b/...in/java/org/apache/skywalking/apm/plugin/kafka/define/Kafka37ConsumerInstrumentation.java
diff --git a/...main/java/org/apache/skywalking/apm/plugin/kafka/define/KafkaConsumerInstrumentation.java b/...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<MethodDescription> 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<MethodDescription> 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<MethodDescription> 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
@@ -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