@Override
    public void setup(ProfilerPluginSetupContext context) {
        final KafkaConfig config = new KafkaConfig(context.getConfig());
        if (Boolean.FALSE == config.isEnable()) {
            logger.info("{} disabled", this.getClass().getSimpleName());
            return;
        }
        logger.info("{} config:{}", this.getClass().getSimpleName(), config);

        if (config.isStreamsEnable()) {
            TransformCallbackParameters parameters = TransformCallbackParametersBuilder.newBuilder()
                    .addBoolean(config.isTraceStreamProcess())
                    .toParameters();
            transformTemplate.transform("org.apache.kafka.streams.processor.internals.StreamTask", StreamTaskTransform.class, parameters);
            transformTemplate.transform("org.apache.kafka.streams.processor.internals.RecordCollectorImpl", RecordCollectorTransform.class);
            transformTemplate.transform("org.apache.kafka.streams.processor.internals.RecordDeserializer", RecordDeserializerTransform.class);
            transformTemplate.transform("org.apache.kafka.streams.processor.internals.StampedRecord", StampedRecordTransform.class);
        }
        if (config.isProducerEnable()) {
            transformTemplate.transform("org.apache.kafka.clients.producer.KafkaProducer", KafkaProducerTransform.class);
            transformTemplate.transform("org.apache.kafka.clients.producer.internals.TransactionManager", TransactionManagerTransform.class);
        }
        if (enableConsumerTransform(config)) {
            transformTemplate.transform("org.apache.kafka.clients.consumer.KafkaConsumer", KafkaConsumerTransform.class);
            transformTemplate.transform("org.apache.kafka.clients.consumer.ConsumerRecord", ConsumerRecordTransform.class);
            // for getting local addresses
            transformTemplate.transform("org.apache.kafka.common.network.Selector", KafkaSelectorTransform.class);
            transformTemplate.transform("org.apache.kafka.clients.NetworkClient", NetworkClientTransform.class);
            transformTemplate.transform("org.apache.kafka.common.TopicPartition", TopicPartitionTransform.class);
            transformTemplate.transform("org.apache.kafka.clients.consumer.ConsumerRecords", ConsumerRecordsTransform.class);
            transformTemplate.transform("org.apache.kafka.common.requests.FetchResponse", FetchResponseTransform.class);
            if (config.isSpringConsumerEnable()) {
                if (config.isKafkaMessageListenerContainerEnable()) {
                    // KafkaMessageListenerContainer$ListenerConsumer
                    transformTemplate.transform("org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer", ListenerConsumerTransform.class);
                }
                transformTemplate.transform("org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter", AcknowledgingConsumerAwareMessageListenerTransform.class);
                transformTemplate.transform("org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter", BatchMessagingMessageListenerAdapterTransform.class);
                // Spring Cloud Starter Stream Kafka 2.2.x is supported
                transformTemplate.transform("org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter", AcknowledgingConsumerAwareMessageListenerTransform.class);
                // for MessagingGatewaySupport in spring-integration-kafka
                transformTemplate.transform("org.springframework.integration.kafka.inbound.KafkaInboundGateway$IntegrationRecordMessageListener", AcknowledgingConsumerAwareMessageListenerTransform.class);
                // for MessageProducerSupport in spring-integration-kafka
                transformTemplate.transform("org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener", AcknowledgingConsumerAwareMessageListenerTransform.class);
                transformTemplate.transform("org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationBatchMessageListener", BatchMessagingMessageListenerAdapterTransform.class);
            }
            if (StringUtils.hasText(config.getKafkaEntryPoint())) {
                transformEntryPoint(config.getKafkaEntryPoint());
            }
        }
    }
@author Harris Gwag (gwagdalf)
@author Taejin Koo
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java