Pop compose changes, polish the code and modify docker image

.github/workflows/test.yml

@@ -69,6 +69,7 @@ jobs:
           - name: zipkin-collector-activemq
           - name: zipkin-collector-kafka
           - name: zipkin-collector-rabbitmq
+          - name: zipkin-collector-pulsar
           - name: zipkin-storage-cassandra
           - name: zipkin-storage-elasticsearch
           - name: zipkin-storage-mysql-v1

docker/examples/README.md

@@ -118,6 +118,18 @@ $ docker compose -f docker-compose-rabbitmq.yml up
 Then configure the [RabbitMQ sender](https://github.com/openzipkin/zipkin-reporter-java/blob/master/amqp-client/src/main/java/zipkin2/reporter/amqp/RabbitMQSender.java)
 using a `host` value of `localhost` or a non-local hostname if in docker.
 
+
+## Pulsar
+
+You can collect traces from [Pulsar](../test-images/zipkin-pulsar/README.md) in addition to HTTP, using the
+`docker-compose-pulsar.yml` file. This configuration starts `zipkin` and `zipkin-pulsar` in their
+own containers.
+
+To add Pulsar configuration, run:
+```bash
+$ docker compose -f docker-compose-pulsar.yml up
+```
+
 ## Eureka
 
 You can register Zipkin for service discovery in [Eureka](../test-images/zipkin-eureka/README.md)

docker/examples/docker-compose-pulsar.yml

@@ -0,0 +1,32 @@
+#
+# Copyright The OpenZipkin Authors
+# SPDX-License-Identifier: Apache-2.0
+#
+
+# This file uses the version 2 docker compose file format, described here:
+# https://docs.docker.com/compose/compose-file/#version-2
+#
+# It extends the default configuration from docker-compose.yml to add a test
+# pulsar server, which is used as a span transport.
+
+version: '2.4'
+
+services:
+  pulsar:
+    image: ghcr.io/openzipkin/zipkin-pulsar:${TAG:-latest}
+    container_name: pulsar
+    ports: # expose the pulsar port so apps can publish spans.
+      - "6650:6650"
+      # - "8080:8080" # uncomment to expose the pulsar http port.
+
+  zipkin:
+    extends:
+      file: docker-compose.yml
+      service: zipkin
+    # slim doesn't include Pulsar support, so switch to the larger image
+    image: ghcr.io/openzipkin/zipkin:${TAG:-latest}
+    environment:
+      - PULSAR_SERVICE_URL=pulsar://pulsar:6650
+    depends_on:
+      pulsar:
+        condition: service_healthy

zipkin-collector/pulsar/src/main/java/zipkin2/collector/pulsar/PulsarSpanConsumer.java

@@ -50,8 +50,8 @@ public void startConsumer() throws PulsarClientException {
         .topic(topic)
         .subscriptionType(SubscriptionType.Shared)
         .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
-        .messageListener(new ZipkinMessageListener<>(collector, metrics))
         .loadConf(consumerProps)
+        .messageListener(new ZipkinMessageListener<>(collector, metrics))
         .subscribe();
   }
 
@@ -66,6 +66,10 @@ public void startConsumer() throws PulsarClientException {
     }
   }
 
+  /**
+   * A message listener implementation for processing messages in a Pulsar consumer,
+   * and it should not be overridden by loadConf as it ensures that zipkin could handle span correctly.
+   */
   record ZipkinMessageListener<T>(Collector collector, CollectorMetrics metrics) implements MessageListener<T> {
 
     @Override public void received(Consumer<T> consumer, Message<T> msg) {

zipkin-collector/pulsar/src/test/java/zipkin2/collector/pulsar/PulsarExtension.java

@@ -57,7 +57,7 @@ PulsarCollector.Builder newCollectorBuilder(String topic) {
 
   static final class PulsarContainer extends GenericContainer<PulsarContainer> {
     PulsarContainer() {
-      super(parse("apachepulsar/pulsar:4.0.2"));
+      super(parse("ghcr.io/openzipkin/zipkin-pulsar:3.4.3"));
       withExposedPorts(BROKER_PORT, BROKER_HTTP_PORT);
       String cmd = "/pulsar/bin/apply-config-from-env.py /pulsar/conf/standalone.conf " +
           "&& bin/pulsar standalone " +