Apache pulsar - LoggingMetricExporter logs trace/span IDs but not getting them in log4j2 logs (Javaagent instrumentation)

[kreiger1234]

Hi,

Been struggling with this. To make sure I am not losing my mind, I created a sample app and am running the pulsar broker within my own container. So entire flow from start to end is pretty basic and controlled by me.

I have a Otel javaagent instrumented application (no spring) that publishes a simple msg to pulsar topic, and another app (again no spring) that consumes from it.

Focusing only on the consumer, when debugging, I see that in the message properties, the trace ids and span ids are populated along with other flags as can be seen below -

In console as well I can see that LoggingMetricExporter from Otel logs that -

Upon further debugging, I also noticed that "OpenTelemetryContextDataProvider" gets called, but when it calls Context.current(), that is blank. I am unfortunately not able to debug past this point.

Some environment details -

Library	Version	Additional Comments
OTEL Java Agent	2.26.1
Pulsar client	2.11.1	(Current client version 4.2.0, there's another major issue with 4.2.0 that OpenTelemetryContextDataProvider does not get called. That appears to be a bug on otel side I think.)
Pulsar broker version	4.2.0	(Since messaging happens fine, It should be unrelated what I use here)
Java	21.0.8

Subscriber code below (Pretty barebones) -

package org.example;

import lombok.SneakyThrows;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.pulsar.client.api.*;


public class PulsarOtelDemoSub {
    public static final String CONNURL = PulsarOtelDemoPub.CONNURL;
    public static final String TOPIC = PulsarOtelDemoPub.TOPIC;
    private static final Logger log = LogManager.getLogger();

    @SneakyThrows
    public static void main(String[] args) {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(CONNURL)
                .build();
        Consumer consumer = client.newConsumer()
                .topic(TOPIC)
                .subscriptionName("my-subscription")
                .subscribe();
        while(true) {
            Message msg = consumer.receive();
            try {
                // Do something with the message
                log.info("Message received: " + new String(msg.getData()));
                // Acknowledge the message
                consumer.acknowledge(msg);
            } catch (Exception e) {
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        }
    }
}

Maven POM (A few other dependencies were added for me to debug by backtracking the codebase) -

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>Test</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>21</maven.compiler.source>
        <maven.compiler.target>21</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <pulsar.version>4.2.0</pulsar.version>
    </properties>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>io.opentelemetry</groupId>
                <artifactId>opentelemetry-bom</artifactId>
                <version>1.60.1</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-bom</artifactId>
                <version>2.25.4</version>
                <scope>import</scope>
                <type>pom</type>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-api</artifactId>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-sdk</artifactId>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-exporter-logging</artifactId>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-sdk-metrics</artifactId>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-exporter-otlp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.pulsar</groupId>
            <artifactId>pulsar-client</artifactId>
            <version>2.11.1</version>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry.instrumentation</groupId>
            <artifactId>opentelemetry-log4j-appender-2.17</artifactId>
            <version>2.26.1-alpha</version>
        </dependency>
        <dependency>
            <groupId>io.opentelemetry.instrumentation</groupId>
            <artifactId>opentelemetry-log4j-context-data-2.17-autoconfigure</artifactId>
            <version>2.26.1-alpha</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.25.4</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.44</version>
        </dependency>
    </dependencies>


</project>

Any ideas or tips on what I can check that I am doing wrong? When OpenTelemetryContextDataProvider is able to get the span from context, I think the entire flow would work fine, but not really sure what's going wrong.

I am not sure if the pulsar jar being shaded has any bearing here.

[kreiger1234]

I completely forgot to mention, the reason I was testing with older pulsar client was because of what the reporter mentioned in this - #8859

[kreiger1234]

So I've been debugging more into the pulsar instrumentation library, and I (may be wrong but) came across some fact that the consumer span seemed to call InstrumentUtil.startAndEnd(...). Honestly the debugging has been a massive pain because classes are just not present on classpath for me to debug, and whatever I try to add, don't seem to match the bytecode, but judging by the name, it hit me that Puslar provides a listener mechanism via the messageListener.

That brought together a chain of thought that the span may be ending after message has been received in current code i.e. before the log, and if I attach a listener and then log in that listener instead of receiving manually, the consumer may still be live (I don't know if that made sense to whoever reads this).

And that seemed to work. I got the trace/span Ids now.

Full sample code for reference -

package org.example;

import lombok.SneakyThrows;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.pulsar.client.api.*;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;


public class PulsarOtelDemoSub {
    public static final String CONNURL = PulsarOtelDemoPub.CONNURL;
    public static final String TOPIC = PulsarOtelDemoPub.TOPIC;
    private static final Logger log = LogManager.getLogger();

    @SneakyThrows
    public static void main(String[] args) {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(CONNURL)
                .build();
        Consumer consumer = client.newConsumer()
                .topic(TOPIC)
                .subscriptionName("my-subscription")
                .messageListener(new MessageListener<byte[]>() {
                    @Override
                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                        log.info("Message received: " + new String(msg.getData(), StandardCharsets.UTF_8));
                    }
                })
                .subscribe();
    }
}

The bit about OpenTelemetryContextDataProvider, well I am suspecting now that's just a debugging issue. The class is present on ThreadContextDataInjector's contextDataProviders list. It's just that the debug point does not hit for some reason unless I downgrade the pulsar client. Not sure if there's some conflict with dependencies or something else. Enough debugging for one day for me :).

Hope it helps out someone else.

[laurit]

Listener based apis work better for tracing because the message processing boundary is well define. We can extract the tracing context from the message and create a span that encompasses the listener method. Spans created inside that method will be children on the processing span created by us and the trace id propagated with the message is observable within the listener method. In contrast when you use an api like Message msg = consumer.receive() there is no way for us to tell what is going to happen with the message next and when you will be done processing it. That is why we can only create a span for the receive call and the trace id won't be observable in the subsequent code. We currently don't have a way to handle this nicely without changing code. One way would be to build a library like in open-telemetry/opentelemetry-java-contrib#1855 that turns the receive calls into something that is easier to instrument. Or you could manually extract the context from the message and activate that where you wish to observe the propagated trace id. This would give you access to the remote producer span not the span created for the receive call.