private TraceId populateTraceIdFromHeaders(TraceContext traceContext, org.apache.kafka.common.header.Headers headers) {
                String transactionId = null;
                String spanID = null;
                String parentSpanID = null;
                String flags = null;
                for (org.apache.kafka.common.header.Header header : headers.toArray()) {
                    if (header.key().equals(Header.HTTP_TRACE_ID.toString())) {
                        transactionId = BytesUtils.toString(header.value());
                    } else if (header.key().equals(Header.HTTP_PARENT_SPAN_ID.toString())) {
                        parentSpanID = BytesUtils.toString(header.value());
                    } else if (header.key().equals(Header.HTTP_SPAN_ID.toString())) {
                        spanID = BytesUtils.toString(header.value());
                    } else if (header.key().equals(Header.HTTP_FLAGS.toString())) {
                        flags = BytesUtils.toString(header.value());
                    }
                }

                if (transactionId == null || spanID == null || parentSpanID == null || flags == null) {
                    return null;
                }

                TraceId traceId = traceContext.createTraceId(transactionId, Long.parseLong(parentSpanID), Long.parseLong(spanID), Short.parseShort(flags));
                return traceId;
            }
@author Taejin Koo
@author Victor.Zxy
agent-module/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/interceptor/ConsumerRecordEntryPointInterceptor.java