Feature/repairkafka3.7

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/Kafka37ConsumerInterceptor.java

@@ -0,0 +1,57 @@
+/*
+ * 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<TopicPartition, List<ConsumerRecord<?, ?>>> fetchRecords(Object retObj) {
+        Map<TopicPartition, List<ConsumerRecord<?, ?>>> 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<ConsumerRecord<?, ?>> recordsForTp = rsp.computeIfAbsent(topicPartition, k -> new ArrayList<>());
+                recordsForTp.add(consumerRecord);
+            }
+        }
+        return rsp;
+    }
+
+}

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.
+ *
+ * <pre>
+ *  1. Record the topic when the client invoke <code>subscribed</code> method
+ *  2. Create the entry span when the client invoke the method <code>pollOnce</code>.
+ *  3. Extract all the <code>Trace Context</code> by iterate all <code>ConsumerRecord</code>
+ *  4. Stop the entry span when <code>pollOnce</code> method finished.
+ * </pre>
+ */
+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);
+    }
+}

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.
+ *
+ * <pre>
+ *  1. Record the topic when the client invoke <code>subscribed</code> method
+ *  2. Create the entry span when the client invoke the method <code>pollOnce</code>.
+ *  3. Extract all the <code>Trace Context</code> by iterate all <code>ConsumerRecord</code>
+ *  4. Stop the entry span when <code>pollOnce</code> method finished.
+ * </pre>
+ */
+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);
+    }
+}

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

@@ -63,103 +63,103 @@ public class KafkaConsumerInstrumentation extends AbstractKafkaInstrumentation {
     @Override
     public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
         return new ConstructorInterceptPoint[] {
-            new ConstructorInterceptPoint() {
-                @Override
-                public ElementMatcher<MethodDescription> getConstructorMatcher() {
-                    return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE);
-                }
+                new ConstructorInterceptPoint() {
+                    @Override
+                    public ElementMatcher<MethodDescription> getConstructorMatcher() {
+                        return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_TYPE);
+                    }
 
-                @Override
-                public String getConstructorInterceptor() {
-                    return CONSUMER_CONFIG_CONSTRUCTOR_INTERCEPTOR_CLASS;
-                }
-            },
-              new ConstructorInterceptPoint() {
-                  @Override
-                  public ElementMatcher<MethodDescription> 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<MethodDescription> 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<MethodDescription> getMethodsMatcher() {
-                    // targeting Kafka Client < 3.2
-                    return named(ENHANCE_METHOD).or(named(ENHANCE_COMPATIBLE_METHOD).and(returns(Map.class)));
-                }
+                new InstanceMethodsInterceptPoint() {
+                    @Override
+                    public ElementMatcher<MethodDescription> 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<MethodDescription> 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<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 String getMethodsInterceptor() {
+                        return INTERCEPTOR_CLASS_KAFKA3_2;
+                    }
 
-                @Override
-                public boolean isOverrideArgs() {
-                    return false;
-                }
-            },
-            new InstanceMethodsInterceptPoint() {
-                @Override
-                public ElementMatcher<MethodDescription> getMethodsMatcher() {
-                    return named(SUBSCRIBE_METHOD)
-                      .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_NAME));
-                }
+                    @Override
+                    public boolean isOverrideArgs() {
+                        return false;
+                    }
+                },
+                new InstanceMethodsInterceptPoint() {
+                    @Override
+                    public ElementMatcher<MethodDescription> 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<MethodDescription> getMethodsMatcher() {
-                    return named(SUBSCRIBE_METHOD)
-                      .and(takesArgumentWithType(0, SUBSCRIBE_INTERCEPT_TYPE_PATTERN));
-                }
+                    @Override
+                    public boolean isOverrideArgs() {
+                        return false;
+                    }
+                },
+                new InstanceMethodsInterceptPoint() {
+                    @Override
+                    public ElementMatcher<MethodDescription> 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<MethodDescription> getMethodsMatcher() {
@@ -184,4 +184,4 @@ public boolean isOverrideArgs() {
     protected ClassMatch enhanceClass() {
         return byName(ENHANCE_CLASS);
     }
-}
+}

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/resources/skywalking-plugin.def

@@ -18,4 +18,6 @@ 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
+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