@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);

            final KafkaConfig config = new KafkaConfig(instrumentor.getProfilerConfig());
            final String methodName = toMethodName(config.getKafkaEntryPoint());
            for (InstrumentMethod method : target.getDeclaredMethods(MethodFilters.name(methodName))) {
                try {
                    String[] parameterTypes = method.getParameterTypes();
                    if (parameterTypes == null) {
                        continue;
                    }

                    for (int i = 0; i < parameterTypes.length; i++) {
                        String parameterType = parameterTypes[i];

                        if (KafkaConstants.CONSUMER_RECORD_CLASS_NAME.equals(parameterType)) {
                            method.addInterceptor(ConsumerRecordEntryPointInterceptor.class, va(i));
                            break;
                        } else if (KafkaConstants.CONSUMER_MULTI_RECORD_CLASS_NAME.equals(parameterType)) {
                            method.addInterceptor(ConsumerMultiRecordEntryPointInterceptor.class, va(i));
                            break;
                        }
                    }
                } catch (Exception e) {
                    if (logger.isWarnEnabled()) {
                        logger.warn("Unsupported method " + method, e);
                    }
                }
            }
            return target.toBytecode();
        }
@author Harris Gwag (gwagdalf)
@author Taejin Koo
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java