Support kafka-clients-3.9.x intercept and Upgrade kafka-clients version in optional-reporter-plugins to 3.9.1 (#780)

diff --git a/CHANGES.md b/CHANGES.md
index 113f769..45b20f1 100644
--- a/CHANGES.md
+++ b/CHANGES.md

@@ -20,6 +20,9 @@
 * Eliminate repeated code with HttpServletRequestWrapper in mvc-annotation-commons.
 * Add the jdk httpclient plugin.
 * Fix Gateway 2.0.x plugin not activated for spring-cloud-starter-gateway 2.0.0.RELEASE.
+* Support kafka-clients-3.9.x intercept.
+* Upgrade kafka-clients version in optional-reporter-plugins to 3.9.1.
+
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/242?closed=1)
 
 ------------------

diff --git a/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/ClassicKafkaConsumerInstrumentation.java b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/ClassicKafkaConsumerInstrumentation.java
new file mode 100644
index 0000000..c2092c6
--- /dev/null
+++ b/apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/define/ClassicKafkaConsumerInstrumentation.java

@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+/**
+ * For Kafka 3.9.x change
+ *
+ * <pre>
+ *  1. The class named <code>LegacyKafkaConsumer</code> was rename to <code>ClassicKafkaConsumer</code>
+ *  2. Because of the enhance class was changed, so we should create new Instrumentation to enhance the new class
+ * </pre>
+ */
+public class ClassicKafkaConsumerInstrumentation extends KafkaConsumerInstrumentation {
+    private static final String ENHANCE_CLASS_39_CLASSIC = "org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer";
+
+    @Override
+    protected ClassMatch enhanceClass() {
+        return byName(ENHANCE_CLASS_39_CLASSIC);
+    }
+}

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 a85deb2..bb98b86 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

@@ -21,3 +21,4 @@
 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
+kafka-3.9.x=org.apache.skywalking.apm.plugin.kafka.define.ClassicKafkaConsumerInstrumentation
\ No newline at end of file

diff --git a/apm-sniffer/optional-reporter-plugins/pom.xml b/apm-sniffer/optional-reporter-plugins/pom.xml
index 32700b1..260ddc3 100644
--- a/apm-sniffer/optional-reporter-plugins/pom.xml
+++ b/apm-sniffer/optional-reporter-plugins/pom.xml

@@ -41,7 +41,7 @@
         <ant-contrib.version>1.0b3</ant-contrib.version>
         <ant-nodeps.version>1.8.1</ant-nodeps.version>
 
-        <kafka-clients.version>2.4.1</kafka-clients.version>
+        <kafka-clients.version>3.9.1</kafka-clients.version>
         <spring-kafka-test.version>2.4.6.RELEASE</spring-kafka-test.version>
     </properties>
 

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 7d8933e..c5a5bfd 100644
--- a/docs/en/setup/service-agent/java-agent/Plugin-list.md
+++ b/docs/en/setup/service-agent/java-agent/Plugin-list.md

@@ -57,6 +57,7 @@
 - jetty-server-9.x
 - kafka-0.11.x/1.x/2.x
 - kafka-3.7.x
+- kafka-3.9.x
 - kotlin-coroutine
 - lettuce-common
 - lettuce-5.x-6.4.x

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 23c5680..63369af 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 @@
 * 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.7.1
+  * [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.9.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

diff --git a/test/plugin/scenarios/kafka-scenario/support-version.list b/test/plugin/scenarios/kafka-scenario/support-version.list
index 6e83b29..f59368c 100644
--- a/test/plugin/scenarios/kafka-scenario/support-version.list
+++ b/test/plugin/scenarios/kafka-scenario/support-version.list

@@ -31,3 +31,4 @@
 3.6.0
 3.7.0
 3.7.1
+3.9.1
\ No newline at end of file