Manual Kafka Distribution Tracing NestJs

[AtharHaider1]

HI,
Could someone suggest me what I'm doing wrong?
I'm running a Confluent Kafka where I'm producing and consuming messages.
I want to do distribution tracing between multiple Microservices . Till now I am able to inject the traceparent into Kafka Producer header and able to get that Header into Consumer microservice , but cant link the newly created span with the Traceparent which I received from Kafka message header generated by Producer
In my consumer I want to handle the message and extracting that TraceId from the header. So far so good this is working, but when I want to create a new Span from there with the context/TraceId as a parent it's not working as expected.

can Someone help me with findings what I am doing wrong

In Producer , Inject Value is working fine
const propagator = new W3CTraceContextPropagator();
const activeContext = {};
propagator.inject(context.active(), activeContext, defaultTextMapSetter);

but in consumer with below code i am unable to set context, every time b3propagation is blank object
const propagator = new W3CTraceContextPropagator();
let input = {traceparent:'',tracestate:''};
const b3header= JSON.parse(headers['correlation-context'].toString());
const b3propagation = propagator.extract(api.context.active(),b3header ,api.defaultTextMapGetter);

const opentelemetry = require("@opentelemetry/sdk-node");
import * as opentelemetryapi from '@opentelemetry/api';
const { registerInstrumentations } = require('@opentelemetry/instrumentation');
const { Resource } = require("@opentelemetry/resources");
const { SemanticResourceAttributes } = require("@opentelemetry/semantic-conventions");
const { PeriodicExportingMetricReader } = require("@opentelemetry/sdk-metrics");
const { OTLPMetricExporter } = require("@opentelemetry/exporter-metrics-otlp-grpc");
const { BatchSpanProcessor,BasicTracerProvider, ConsoleSpanExporter, SimpleSpanProcessor } = require('@opentelemetry/sdk-trace-base');
const { OTLPTraceExporter } = require('@opentelemetry/exporter-trace-otlp-grpc');
const { AWSXRayPropagator } = require("@opentelemetry/propagator-aws-xray");
const { AWSXRayIdGenerator } = require("@opentelemetry/id-generator-aws-xray");
const { HttpInstrumentation } = require("@opentelemetry/instrumentation-http");
const { AwsInstrumentation } = require("@opentelemetry/instrumentation-aws-sdk");
const { KafkaJsInstrumentation } = require('opentelemetry-instrumentation-kafkajs');
const { NodeTracerProvider } = require('@opentelemetry/sdk-trace-node');
import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
import {CompositePropagator, W3CTraceContextPropagator, W3CBaggagePropagator, } from '@opentelemetry/core';
import { B3InjectEncoding, B3Propagator } from '@opentelemetry/propagator-b3';

opentelemetryapi.propagation.setGlobalPropagator(new B3Propagator());

const _resource = Resource.default().merge(new Resource({
[SemanticResourceAttributes.SERVICE_NAME]: "PublishMxService",
[SemanticResourceAttributes.SERVICE_NAMESPACE]: 'testing',
[SemanticResourceAttributes.SERVICE_VERSION]: '0.0.1-beta',
[SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: 'development',
}));

const tracerProvider = new NodeTracerProvider({
// be sure to disable old plugin
plugins: {
    kafkajs: { enabled: false, path: 'opentelemetry-plugin-kafkajs' }
}
});

const _traceExporter = new OTLPTraceExporter();
const _spanProcessor = new BatchSpanProcessor(_traceExporter);

const _tracerConfig = {
idGenerator: new AWSXRayIdGenerator(),
}
const _metricReader = new PeriodicExportingMetricReader({
exporter: new OTLPMetricExporter(),
exportIntervalMillis: 1000
});

async function nodeSDKBuilder()
{
const contextManager = new AsyncHooksContextManager();
contextManager.enable();
const sdk = new opentelemetry.NodeSDK({
tracerProvider,
textMapPropagator: new CompositePropagator({
propagators: [
new W3CTraceContextPropagator(),
new B3Propagator({
injectEncoding: B3InjectEncoding.MULTI_HEADER,
}),
new AWSXRayPropagator(),
],
}),

    metricReader: _metricReader,
    instrumentations: [
        new HttpInstrumentation(),
        new AwsInstrumentation({
            suppressInternalInstrumentation: true
        }),
        new KafkaJsInstrumentation({
            // see kafkajs instrumentation docs for available configuration
          })
    ],
    resource: _resource,
    spanProcessor: _spanProcessor,
    traceExporter: _traceExporter,
});

registerInstrumentations({
  instrumentations: [
    
    new KafkaJsInstrumentation()
  ],

});

sdk.configureTracerProvider(_tracerConfig, _spanProcessor);

// this enables the API to record telemetry
await sdk.start(); 

// gracefully shut down the SDK on process exit
process.on('SIGTERM', () => {
sdk.shutdown()
  .then(() => console.log('Tracing and Metrics terminated'))
  .catch((error) => console.log('Error terminating tracing and metrics', error))
  .finally(() => process.exit(0));
});

}
export default { nodeSDKBuilder };