Fix kafka plugin sometimes trace doesn't show span layer (#5264)

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackAdapterInterceptor.java

@@ -24,6 +24,7 @@ import org.apache.skywalking.apm.agent.core.context.ContextManager;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.tag.Tags;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 
 /**
@@ -44,6 +45,7 @@ public class CallbackAdapterInterceptor implements Callback {
     public void onCompletion(RecordMetadata metadata, Exception exception) {
         ContextSnapshot snapshot = callbackCache.getSnapshot();
         AbstractSpan activeSpan = ContextManager.createLocalSpan("Kafka/Producer/Callback");
+        SpanLayer.asMQ(activeSpan);
         activeSpan.setComponent(ComponentsDefine.KAFKA_PRODUCER);
         if (metadata != null) {
             Tags.MQ_TOPIC.set(activeSpan, metadata.topic());

apm-sniffer/apm-sdk-plugin/kafka-plugin/src/main/java/org/apache/skywalking/apm/plugin/kafka/CallbackInterceptor.java

@@ -23,6 +23,7 @@ import org.apache.skywalking.apm.agent.core.context.ContextManager;
 import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
 import org.apache.skywalking.apm.agent.core.context.tag.Tags;
 import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
+import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
@@ -43,6 +44,7 @@ public class CallbackInterceptor implements InstanceMethodsAroundInterceptor {
             ContextSnapshot snapshot = getSnapshot(cache);
             RecordMetadata metadata = (RecordMetadata) allArguments[0];
             AbstractSpan activeSpan = ContextManager.createLocalSpan("Kafka/Producer/Callback");
+            SpanLayer.asMQ(activeSpan);
             activeSpan.setComponent(ComponentsDefine.KAFKA_PRODUCER);
             if (metadata != null) {
                 // Null if an error occurred during processing of this record

test/plugin/scenarios/kafka-scenario/config/expectedData.yaml

@@ -23,7 +23,7 @@ segmentItems:
       operationId: 0
       parentSpanId: -1
       spanId: 0
-      spanLayer: Unknown
+      spanLayer: MQ
       startTime: nq 0
       endTime: nq 0
       componentId: 40
@@ -43,7 +43,7 @@ segmentItems:
         operationId: 0
         parentSpanId: -1
         spanId: 0
-        spanLayer: Unknown
+        spanLayer: MQ
         startTime: nq 0
         endTime: nq 0
         componentId: 40