@Override
        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.consumer.ConsumerConfig",
                    "org.apache.kafka.common.serialization.Deserializer", "org.apache.kafka.common.serialization.Deserializer");
            if (constructor != null) {
                constructor.addInterceptor(ConsumerConstructorInterceptor.class);
            }

            if (constructor == null) {
                constructor = target.getConstructor("java.util.Map", "org.apache.kafka.common.serialization.Deserializer", "org.apache.kafka.common.serialization.Deserializer");
                if (constructor != null) {
                    constructor.addInterceptor(ConsumerConstructor_V_2_7_Interceptor.class);
                }
            }

            InstrumentMethod pollLongMethod = target.getDeclaredMethod("poll", "long");
            if (pollLongMethod != null) {
                pollLongMethod.addScopedInterceptor(ConsumerPollInterceptor.class, SCOPE_KAFKA_CONSUMER_POLL);
            }
            InstrumentMethod pollDurationMethod = target.getDeclaredMethod("poll", "java.time.Duration");
            if (pollDurationMethod != null) {
                pollDurationMethod.addScopedInterceptor(ConsumerPollInterceptor.class, SCOPE_KAFKA_CONSUMER_POLL);
            }

            target.addField(RemoteAddressFieldAccessor.class);

            return target.toBytecode();
        }
@author Harris Gwag (gwagdalf)
@author Taejin Koo
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java