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

            // Version 2.8.0+ is supported.
            InstrumentMethod constructor = target.getConstructor("org.apache.kafka.clients.producer.ProducerConfig",
                    "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer",
                    "org.apache.kafka.clients.producer.internals.ProducerMetadata", "org.apache.kafka.clients.KafkaClient",
                    "org.apache.kafka.clients.producer.internals.ProducerInterceptors", "org.apache.kafka.common.utils.Time");

            if (constructor == null) {
                // Version 2.3.0+ is supported.
                constructor = target.getConstructor("java.util.Map",
                        "org.apache.kafka.common.serialization.Serializer", "org.apache.kafka.common.serialization.Serializer",
                        "org.apache.kafka.clients.producer.internals.ProducerMetadata", "org.apache.kafka.clients.KafkaClient",
                        "org.apache.kafka.clients.producer.internals.ProducerInterceptors", "org.apache.kafka.common.utils.Time");
            }

            if (constructor == null) {
                // Version 2.2.0+ is supported.
                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) {
                constructor = target.getConstructor("org.apache.kafka.clients.producer.ProducerConfig",
                        "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");
            }

            if (constructor != null) {
                constructor.addInterceptor(ProducerConstructorInterceptor.class);
            }

            InstrumentMethod sendMethod = target.getDeclaredMethod("send", "org.apache.kafka.clients.producer.ProducerRecord", "org.apache.kafka.clients.producer.Callback");
            if (sendMethod != null) {
                sendMethod.addInterceptor(ProducerSendInterceptor.class);
            }

            // Version 0.11.0+ is supported.
            InstrumentMethod setReadOnlyMethod = target.getDeclaredMethod("setReadOnly", "org.apache.kafka.common.header.Headers");
            if (setReadOnlyMethod != null) {
                setReadOnlyMethod.addInterceptor(ProducerAddHeaderInterceptor.class);
            }
            if (target.hasField("apiVersions")) {
                target.addGetter(ApiVersionsGetter.class, "apiVersions");
            }

            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