opentelemetry-instrumentation-confluent-kafka Tracing: Spans Not Connecting

[knut-bw]

My producer and consumer spans aren't linking up. I'm attaching the traceparent to the context and I can retrieve it from the message headers, but the spans still aren't connected. Why is this happening?
package version:
confluent-kafka 2.7.0
opentelemetry-instrumentation-confluent-kafka 0.51b0
This is my producer

resource = Resource(attributes={
SERVICE_NAME: "my-service-name"
})
traceProvider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="xxxxxx", insecure=True))
traceProvider.add_span_processor(processor)
composite_propagator = CompositePropagator([
TraceContextTextMapPropagator(),
W3CBaggagePropagator(),
])
propagate.set_global_textmap(composite_propagator)
trace.set_tracer_provider(traceProvider)
tracer = trace.get_tracer(__name__)
# Kafka Configuration (from environment variables)
KAFKA_BOOTSTRAP_SERVERS = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "xxxxxx")
KAFKA_TOPIC = os.environ.get("KAFKA_TOPIC", "xxxxxx")
KAFKA_GROUP_ID = os.environ.get("KAFKA_GROUP_ID", "emqx_consumer_group")
CREATE_TOPIC = os.environ.get("CREATE_TOPIC", "false").lower() == "true" # Flag to create the topic if it doesn't exist
ConfluentKafkaInstrumentor().instrument()
inst = ConfluentKafkaInstrumentor()
conf1 = {'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS}
producer = Producer(conf1)
p = inst.instrument_producer(producer, tracer_provider=traceProvider)
# Get environment variables for MQTT configuration
MQTT_BROKER = os.environ.get("MQTT_BROKER", "xxxxxxx")
MQTT_PORT = int(os.environ.get("MQTT_PORT", xxxxxx))
MQTT_SUB_TOPIC = os.environ.get("MQTT_TOPIC", "test2")
# MQTT_PUB_TOPIC = os.environ.get("MQTT_TOPIC", "test2s")
CLIENT_ID = os.environ.get("CLIENT_ID", "mqtt-microservice")
def producer_kafka_message():
context_setter = KafkaContextSetter()
new_carrier = {}
new_carrier["tracestate"] = "congo=t61rcWkgMzE" propagate.inject(carrier=new_carrier) kafka_headers = [(key, value.encode("utf-8")) for key, value in new_carrier.items()]
p.produce(topic=KAFKA_TOPIC, value=b'aaaaa', headers=kafka_headers)
p.poll(0)
p.flush()

This is my consumer

ConfluentKafkaInstrumentor().instrument()
inst = ConfluentKafkaInstrumentor()
resource = Resource(attributes={
SERVICE_NAME: "other-service-name"
})
traceProvider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="xxxxxxx", insecure=True))
traceProvider.add_span_processor(processor)
loop = asyncio.get_event_loop()
composite_propagator = CompositePropagator([
TraceContextTextMapPropagator(),
W3CBaggagePropagator(),
])
propagate.set_global_textmap(composite_propagator)
KAFKA_BOOTSTRAP_SERVERS = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "xxxxxxx")
KAFKA_TOPIC = os.environ.get("KAFKA_TOPIC", "test-topic-room1")
KAFKA_GROUP_ID = os.environ.get("KAFKA_GROUP_ID", "emqx_consumer_group")
CREATE_TOPIC = os.environ.get("CREATE_TOPIC", "false").lower() == "true"  # Flag to create the topic if it doesn't exist
conf2 = {
'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
'group.id': KAFKA_GROUP_ID,
'auto.offset.reset': 'latest'
}
# report a span of type consumer with the default settings
consumer = Consumer(conf2)
c = inst.instrument_consumer(consumer, tracer_provider=traceProvider)
consumer.subscribe([KAFKA_TOPIC])
def basic_consume_loop(consumer):
print(f"Consuming messages from topic '{KAFKA_TOPIC}'...")
current_span = trace.get_current_span()
try:
# create_kafka_topic()
while True:
msg = c.poll()
if msg is None:
continue
if msg.error():
print('msg.error()', msg.error())
print("Consumer error: {}".format(msg.error()))
if msg.error().code() == "KafkaError._PARTITION_EOF":
print("msg.error().code()", msg.error().code())
# End of partition event
# print(f"{msg.topic() [{msg.partition()}] reached end at offset {msg.offset()}}")
elif msg.error():
print("msg.error()", msg.error())
# raise KafkaException(msg.error())
headers = {key: value.decode('utf-8') for key, value in msg.headers()}
prop = TraceContextTextMapPropagator()
ctx = prop.extract(carrier=headers)