Repository: pinpoint-apm/pinpoint
Issue: Adds support Kafka 2.2.0+  
Labels: enhancement
module:plugin

#### description 
There are major method changes in version 2.2.0.

* KafkaCosumer poll method
```
# 2.0.0+
org.apache.kafka.clients.consumer.KafkaConsumer#poll(long, boolean)

->

# 2.2.0+
org.apache.kafka.clients.consumer.KafkaConsumer#poll(org.apache.kafka.common.utils.Timer, boolean)
```

* KafkaProducer constructor 
```
org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(org.apache.kafka.clients.producer.ProducerConfig, org.apache.kafka.common.serialization.Serializer<K>, org.apache.kafka.common.serialization.Serializer<V>, org.apache.kafka.clients.Metadata, org.apache.kafka.clients.KafkaClient)

->

# 2.2.0+
org.apache.kafka.clients.producer.KafkaProducer#KafkaProducer(java.util.Map<java.lang.String,java.lang.Object>, org.apache.kafka.common.serialization.Serializer<K>, org.apache.kafka.common.serialization.Serializer<V>, org.apache.kafka.clients.Metadata, org.apache.kafka.clients.KafkaClient, org.apache.kafka.clients.producer.internals.ProducerInterceptors, org.apache.kafka.common.utils.Time)
```

Issue comments:
resolved via #6151 #6155 

Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
index d1c527d52792..dc5cb7e9cbac 100644
--- a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
+++ b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
@@ -55,8 +55,11 @@ public void setup(ProfilerPluginSetupContext context) {
                 public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, String className, Class<?> classBeingRedefined, ProtectionDomain protectionDomain, byte[] classfileBuffer) throws InstrumentException {
                     final InstrumentClass target = instrumentor.getInstrumentClass(classLoader, className, classfileBuffer);
 
-                    InstrumentMethod constructor = target.getConstructor("org.apache.kafka.clients.producer.ProducerConfig",
-                            "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer");
+                    // Version 2.2.0+ is supported.
+                    InstrumentMethod constructor = target.getConstructor("java.util.Map",
+                            "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer",
+                            "org.apache.kafka.clients.Metadata", "org.apache.kafka.clients.KafkaClient",
+                            "org.apache.kafka.clients.producer.internals.ProducerInterceptors", "org.apache.kafka.common.utils.Time");
 
                     // Version 2.0.0+ is supported.
                     if (constructor == null) {
@@ -64,6 +67,12 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader,
                                 "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer",
                                 "org.apache.kafka.clients.Metadata", "org.apache.kafka.clients.KafkaClient");
                     }
+
+                    if (constructor == null) {
+                        constructor = target.getConstructor("org.apache.kafka.clients.producer.ProducerConfig",
+                                "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer");
+                    }
+
                     constructor.addInterceptor(KafkaConstants.PRODUCER_CONSTRUCTOR_INTERCEPTOR);
 
                     InstrumentMethod sendMethod = target.getDeclaredMethod("send", "org.apache.kafka.clients.producer.ProducerRecord", "org.apache.kafka.clients.producer.Callback");
@@ -88,12 +97,19 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader,
                             "org.apache.kafka.common.serialization.Deserializer", "org.apache.kafka.common.serialization.Deserializer");
                     constructor.addInterceptor(KafkaConstants.CONSUMER_CONSTRUCTOR_INTERCEPTOR);
 
+                    // Version 2.2.0+ is supported.
+                    InstrumentMethod pollMethod = target.getDeclaredMethod("poll", "org.apache.kafka.common.utils.Timer", "boolean");
+
                     // Version 2.0.0+ is supported.
-                    InstrumentMethod pollMethod = target.getDeclaredMethod("poll", "long", "boolean");
+                    if (pollMethod == null) {
+                        pollMethod = target.getDeclaredMethod("poll", "long", "boolean");
+                    }
 
+                    // Version 2.0.0-
                     if (pollMethod == null) {
                         pollMethod = target.getDeclaredMethod("poll", "long");
                     }
+
                     pollMethod.addInterceptor(KafkaConstants.CONSUMER_POLL_INTERCEPTOR);
 
                     target.addField(KafkaConstants.REMOTE_ADDRESS_ACCESSOR);
diff --git a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ProducerConstructorInterceptor.java b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ProducerConstructorInterceptor.java
index ffb7b18cab12..2d1cca62cc04 100644
--- a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ProducerConstructorInterceptor.java
+++ b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ProducerConstructorInterceptor.java
@@ -23,15 +23,19 @@
 import com.navercorp.pinpoint.common.util.CollectionUtils;
 import com.navercorp.pinpoint.plugin.kafka.KafkaConstants;
 import com.navercorp.pinpoint.plugin.kafka.field.accessor.RemoteAddressFieldAccessor;
+
 import org.apache.kafka.clients.producer.ProducerConfig;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * @author Taejin Koo
  */
 public class ProducerConstructorInterceptor implements AroundInterceptor {
 
+    private static final String KEY_BOOTSTRAP_SERVERS = "bootstrap.servers";
+
     private final PLogger logger = PLoggerFactory.getLogger(this.getClass());
     private final boolean isDebug = logger.isDebugEnabled();
 
@@ -57,12 +61,21 @@ public void after(Object target, Object[] args, Object result, Throwable throwab
         }
 
         ProducerConfig producerConfig = getProducerConfig(args);
-        if (producerConfig == null) {
+        if (producerConfig != null) {
+            String remoteAddress = getRemoteAddress(producerConfig);
+            ((RemoteAddressFieldAccessor) target)._$PINPOINT$_setRemoteAddress(remoteAddress);
+            return;
+        }
+
+        // Version 2.2.0+ is supported.
+        Map map = getMap(args);
+        if (map != null) {
+            Object remoteAddressObject = map.get(KEY_BOOTSTRAP_SERVERS);
+            String remoteAddress = getRemoteAddress0(remoteAddressObject);
+            ((RemoteAddressFieldAccessor) target)._$PINPOINT$_setRemoteAddress(remoteAddress);
             return;
         }
 
-        String remoteAddress = getRemoteAddress(producerConfig);
-        ((RemoteAddressFieldAccessor) target)._$PINPOINT$_setRemoteAddress(remoteAddress);
     }
 
     private ProducerConfig getProducerConfig(Object args[]) {
@@ -71,20 +84,43 @@ private ProducerConfig getProducerConfig(Object args[]) {
         }
 
         if (args[0] instanceof ProducerConfig) {
-            return (ProducerConfig)args[0];
+            return (ProducerConfig) args[0];
+        }
+
+        return null;
+    }
+
+    private Map getMap(Object args[]) {
+        if (ArrayUtils.isEmpty(args)) {
+            return null;
+        }
+
+        if (args[0] instanceof Map) {
+            return (Map) args[0];
         }
 
         return null;
     }
 
     private String getRemoteAddress(ProducerConfig producerConfig) {
-        List<String> serverList = producerConfig.getList("bootstrap.servers");
+        List<String> serverList = producerConfig.getList(KEY_BOOTSTRAP_SERVERS);
+        return getRemoteAddress0(serverList);
+    }
+
+    private String getRemoteAddress0(Object remoteAddressObject) {
         String remoteAddress = KafkaConstants.UNKNOWN;
-        if (CollectionUtils.nullSafeSize(serverList) == 1) {
-            remoteAddress = serverList.get(0);
-        } else if (CollectionUtils.nullSafeSize(serverList) > 1) {
-            remoteAddress = serverList.toString();
+        if (remoteAddressObject instanceof List) {
+            List remoteAddressList = (List) remoteAddressObject;
+            if (CollectionUtils.nullSafeSize(remoteAddressList) == 1) {
+                remoteAddress = String.valueOf(remoteAddressList.get(0));
+            } else if (CollectionUtils.nullSafeSize(remoteAddressList) > 1) {
+                remoteAddress = remoteAddressList.toString();
+            }
+            return remoteAddress;
+        } else if (remoteAddressObject instanceof String) {
+            return (String) remoteAddressObject;
         }
+
         return remoteAddress;
     }
 


Test patch: