@Override
            public Trace createTrace(TraceContext traceContext, ConsumerRecord<?, ?> consumerRecord) {
                org.apache.kafka.common.header.Headers headers = consumerRecord.headers();
                if (headers == null) {
                    return createTrace0(traceContext, consumerRecord);
                }

                if (!isSampled(headers)) {
                    // Even if this transaction is not a sampling target, we have to create Trace object to mark 'not sampling'.
                    // For example, if this transaction invokes rpc call, we can add parameter to tell remote node 'don't sample this transaction'
                    final Trace trace = traceContext.disableSampling();
                    if (isDebug) {
                        logger.debug("remotecall sampling flag found. skip trace");
                    }
                    return trace;
                }

                TraceId traceId = populateTraceIdFromHeaders(traceContext, headers);
                final Trace trace;
                if (traceId != null) {
                    trace = createContinueTrace(traceContext, consumerRecord, traceId);
                } else {
                    trace = createTrace0(traceContext, consumerRecord);
                }
                if (trace.canSampled() && headerRecorded) {
                    headerRecorder.record(trace.getSpanRecorder(), consumerRecord);
                }
                return trace;
            }
@author Taejin Koo
@author Victor.Zxy
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ConsumerRecordEntryPointInterceptor.java