Fixed test case errors and reviewed comments.

Author: GuiSong01

.github/workflows/plugins-jdk17-test.1.yaml

@@ -65,6 +65,7 @@ jobs:
           - c3p0-0.9.2.x-0.10.x-scenario
           - spring-scheduled-6.x-scenario
           - caffeine-3.x-scenario
+          - spring-kafka-3.3.x-scenario
     steps:
       - uses: actions/checkout@v2
         with:

.github/workflows/plugins-test.3.yaml

@@ -81,7 +81,6 @@ jobs:
           - spring-kafka-1.3.x-scenario
           - spring-kafka-2.2.x-scenario
           - spring-kafka-2.3.x-scenario
-          - spring-kafka-3.3.x-scenario
           - spring-scheduled-3.x-5.x-scenario
           - elasticjob-2.x-scenario
           - quartz-scheduler-2.x-scenario

…a/ExtendedConstructorInterceptPoint.java …afka/ExtendedConstructorInterceptor.javaapm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/ExtendedConstructorInterceptPoint.java renamed to apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/ExtendedConstructorInterceptor.java apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/ExtendedConstructorInterceptPoint.java renamed to apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/ExtendedConstructorInterceptor.java

@@ -23,45 +23,25 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
 
-public class ExtendedConstructorInterceptPoint implements InstanceConstructorInterceptor {
+public class ExtendedConstructorInterceptor implements InstanceConstructorInterceptor {
 
     @Override
     public void onConstruct(final EnhancedInstance objInst, final Object[] allArguments) throws Throwable {
         ExtendedConsumerEnhanceRequiredInfo requiredInfo = new ExtendedConsumerEnhanceRequiredInfo();
-        extractConsumerConfig(allArguments, requiredInfo);
-        objInst.setSkyWalkingDynamicField(requiredInfo);
-    }
 
-    private void extractConsumerConfig(Object[] allArguments, ExtendedConsumerEnhanceRequiredInfo requiredInfo) {
-        if (allArguments == null || allArguments.length == 0) {
-            return;
+        Map<String, Object> configMap = (Map<String, Object>) allArguments[0];
+        Object bootstrapServers = configMap.get("bootstrap.servers");
+        if (bootstrapServers instanceof List) {
+            requiredInfo.setBrokerServers(String.join(";", (List<String>) bootstrapServers));
+        } else if (bootstrapServers != null) {
+            requiredInfo.setBrokerServers(String.valueOf(bootstrapServers).trim().replaceAll("\\s*,\\s*", ";"));
         }
 
-        for (Object arg : allArguments) {
-            if (arg instanceof Map) {
-                extractConfigFromMap(arg, requiredInfo);
-                break;
-            }
+        Object groupId = configMap.get("group.id");
+        if (groupId != null) {
+            requiredInfo.setGroupId(groupId.toString());
         }
-    }
-
-    private void extractConfigFromMap(Object arg, ExtendedConsumerEnhanceRequiredInfo requiredInfo) {
-        try {
-            Map<String, Object> configMap = (Map<String, Object>) arg;
-            Object bootstrapServers = configMap.get("bootstrap.servers");
-            if (bootstrapServers instanceof List) {
-                requiredInfo.setBrokerServers(String.join(";", (List<String>) bootstrapServers));
-            } else if (bootstrapServers != null) {
-                requiredInfo.setBrokerServers(String.valueOf(bootstrapServers).trim().replaceAll("\\s*,\\s*", ";"));
-            }
 
-            Object groupId = configMap.get("group.id");
-            if (groupId != null) {
-                requiredInfo.setGroupId(groupId.toString());
-            }
-        } catch (Exception e) {
-            // Ignore exception and continue
-        }
+        objInst.setSkyWalkingDynamicField(requiredInfo);
     }
-
 }

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/kafka/define/ExtendedKafkaConsumerInstrumentation.java

@@ -25,15 +25,15 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
 import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
 
-import static net.bytebuddy.matcher.ElementMatchers.any;
 import static net.bytebuddy.matcher.ElementMatchers.named;
+import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
 import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
 
 /**
  * Enhanced the ExtendedKafkaConsumer class to intercept poll method
  */
 public class ExtendedKafkaConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
-
+    public static final String CONSTRUCTOR_INTERCEPT_MAP_TYPE = "java.util.Map";
     private static final String ENHANCE_CLASS = "org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer";
     private static final String CONSTRUCTOR_INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.spring.kafka.ExtendedConstructorInterceptPoint";
     private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.spring.kafka.ExtendedKafkaConsumerInterceptor";
@@ -44,7 +44,7 @@ public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
             new ConstructorInterceptPoint() {
                 @Override
                 public ElementMatcher<MethodDescription> getConstructorMatcher() {
-                    return any();
+                    return takesArgumentWithType(0, CONSTRUCTOR_INTERCEPT_MAP_TYPE);
                 }
 
                 @Override

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-kafka-2.x-3.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/kafka/ExtendedKafkaConsumerInterceptorTest.java

@@ -155,17 +155,9 @@ public void testAfterMethodWithRecords() throws Throwable {
         assertThat(span.getOperationName(), is("Kafka/test/Consumer/testGroup"));
     }
 
-    @Test
-    public void testOnConstructWithEmptyArguments() throws Throwable {
-        ExtendedConstructorInterceptPoint constructorInterceptPoint = new ExtendedConstructorInterceptPoint();
-        constructorInterceptPoint.onConstruct(enhancedInstance, new Object[] {});
-        assertThat(requiredInfo.getGroupId(), is("Unknown"));
-        assertThat(requiredInfo.getBrokerServers(), is("Unknown"));
-    }
-
     @Test
     public void testOnConstructWithMapConfig() throws Throwable {
-        ExtendedConstructorInterceptPoint constructorInterceptPoint = new ExtendedConstructorInterceptPoint();
+        ExtendedConstructorInterceptor constructorInterceptPoint = new ExtendedConstructorInterceptor();
         Map<String, Object> config = new HashMap<>();
         config.put("bootstrap.servers", "localhost:9092");
         config.put("group.id", "testGroup");

test/plugin/scenarios/spring-kafka-3.3.x-scenario/pom.xml

@@ -27,7 +27,7 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <compiler.version>1.8</compiler.version>
+        <compiler.version>17</compiler.version>
         <maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version>
         <test.framework.version>3.3.10</test.framework.version>
         <log4j.version>2.6.2</log4j.version>