Pulsar Traces are not always propagated

[michalcukierman]

This is a regression introduced by #2456.
Tracing properties from incoming Pulsar messages are not propagated to outgoing Pulsar messages.

The change is in PulsarTrace.java:

before:

        public Builder withProperties(Map<String, String> properties) {
            this.properties = properties;
            return this;
        }

after:

        public Builder withProperties(Map<String, String> properties) {
            if (properties != null) {
                this.properties = new HashMap<>(properties);
            }
            return this;
        }

It was introduced to ensure that the map is modifiable because it's required during producing outgoing traces.

Unfortunately, it introduced a regression in PulsarOutgoingChannel. We expect that the changes in properties will be propagated into the state of the trace:

            if (tracingEnabled) {
                Map<String, String> properties = new HashMap<>(); // Single instance has to be used in Trace and MessageBuilder
                TracingUtils.traceOutgoing(instrumenter, message, new PulsarTrace.Builder()
                        .withProperties(properties) // REGRESSION HERE, we create a new instance of a HashMap
                        .withTopic(producer.getTopic())
                        .build());
                messageBuilder.properties(properties);
            }

I think that the correct solution could look like:

            if (tracingEnabled) {
                PulsarTrace trace = new PulsarTrace.Builder() // Builder initializes properties with new HashMap
                        .withTopic(producer.getTopic())
                        .build();
                TracingUtils.traceOutgoing(instrumenter, message, trace);
                messageBuilder.properties(trace.getProperties()); // Reuse the same instance of the HashMap
            }

This happens only when PulsarOutgoingMessageMetadata is not used.

[michalcukierman]

During the tests, I discovered that more changes are required. Pushed the second commit. In the second case it's also about re-using the same instance of a map between the PulsarTrace and the MessageBuilder.

Please feel free to reject this PR and propose other solutions that would work.