void recordRootSpan(SpanRecorder recorder, ConsumerRecord<?, ?> consumerRecord, String parentApplicationName, String parentApplicationType) {
                recorder.recordServiceType(KafkaConstants.KAFKA_CLIENT);
                recorder.recordApi(ConsumerRecordEntryPointInterceptor.ENTRY_POINT_METHOD_DESCRIPTOR);

                String endPointAddress = getEndPointAddress(consumerRecord);
                String remoteAddress = getRemoteAddress(consumerRecord);
                endPointAddress = StringUtils.defaultIfEmpty(endPointAddress, remoteAddress);

                recorder.recordEndPoint(endPointAddress);
                recorder.recordRemoteAddress(remoteAddress);

                String topic = consumerRecord.topic();
                recorder.recordRpcName(createRpcName(consumerRecord));
                recorder.recordAcceptorHost(remoteAddress);
                recorder.recordAttribute(KafkaConstants.KAFKA_TOPIC_ANNOTATION_KEY, topic);

                recorder.recordAttribute(KafkaConstants.KAFKA_PARTITION_ANNOTATION_KEY, consumerRecord.partition());
                recorder.recordAttribute(KafkaConstants.KAFKA_OFFSET_ANNOTATION_KEY, consumerRecord.offset());

                if (StringUtils.hasText(parentApplicationName) && StringUtils.hasText(parentApplicationType)) {
                    recorder.recordParentApplication(parentApplicationName, NumberUtils.parseShort(parentApplicationType, ServiceType.UNDEFINED.getCode()));
                }
            }
@author Taejin Koo
@author Victor.Zxy
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ConsumerRecordEntryPointInterceptor.java