Adds Pulsar sender

README.md

@@ -7,8 +7,8 @@
 Zipkin Reporter buffers and sends trace data collected from tracer libraries to a Zipkin compatible backend.
 
 This repository includes a Java reporting library with transport-specific senders.
-Transport options include HTTP, Apache ActiveMQ, Apache Kafka, gRPC, RabbitMQ
-and Scribe (Apache Thrift). Requires JRE 6 or later.
+Transport options include HTTP, Apache ActiveMQ, Apache Kafka, gRPC, RabbitMQ, Scribe (Apache Thrift)
+and Apache Pulsar. Requires JRE 6 or later.
 
 # Usage
 These components can be called when spans have been recorded and ready to send to zipkin.

benchmarks/pom.xml

@@ -109,6 +109,12 @@
       <version>${testcontainers.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-sender-pulsar-client</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

benchmarks/src/test/java/zipkin2/reporter/PulsarSenderBenchmarks.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.reporter;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HttpWaitStrategy;
+import zipkin2.reporter.internal.SenderBenchmarks;
+import zipkin2.reporter.pulsar.PulsarSender;
+
+import java.time.Duration;
+import java.util.Collections;
+
+import static org.testcontainers.utility.DockerImageName.parse;
+
+public class PulsarSenderBenchmarks extends SenderBenchmarks {
+  static final Logger LOGGER = LoggerFactory.getLogger(PulsarSenderBenchmarks.class);
+
+  static final class PulsarContainer extends GenericContainer<PulsarContainer> {
+    static final int BROKER_PORT = 6650;
+    static final int BROKER_HTTP_PORT = 8080;
+
+    PulsarContainer() {
+      super(parse("apachepulsar/pulsar:4.0.2"));
+      withExposedPorts(BROKER_PORT, BROKER_HTTP_PORT);
+      String cmd = "/pulsar/bin/apply-config-from-env.py /pulsar/conf/standalone.conf " +
+          "&& bin/pulsar standalone " +
+          "--no-functions-worker -nss";
+      withEnv("PULSAR_MEM", "-Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g"); // limit memory usage
+      waitStrategy = new HttpWaitStrategy()
+          .forPort(BROKER_HTTP_PORT)
+          .forStatusCode(200)
+          .forPath("/admin/v2/clusters")
+          .withStartupTimeout(Duration.ofSeconds(120));
+      withCommand("/bin/bash", "-c", cmd);
+      withLogConsumer(new Slf4jLogConsumer(LOGGER));
+    }
+
+    String serviceUrl() {
+      return "pulsar://" + getHost() + ":" + getMappedPort(BROKER_PORT);
+    }
+  }
+
+  PulsarContainer pulsar;
+  Consumer<byte[]> consumer;
+
+  @Override protected BytesMessageSender createSender() throws PulsarClientException {
+    pulsar = new PulsarContainer();
+    pulsar.start();
+
+    String topicName = "zipkin";
+    PulsarSender sender = PulsarSender.newBuilder().topic(topicName)
+        .serviceUrl(pulsar.serviceUrl()).build();
+
+    sender.send(Collections.emptyList());
+    try (PulsarClient client = PulsarClient.builder()
+        .serviceUrl(pulsar.serviceUrl())
+        .build()) {
+      consumer = client.newConsumer()
+          .topic(topicName)
+          .subscriptionType(SubscriptionType.Shared)
+          .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+          .subscriptionName(topicName)
+          .subscribe();
+    }
+
+    return sender;
+  }
+
+  @Override protected void afterSenderClose() {
+    pulsar.stop();
+  }
+
+  // Convenience main entry-point
+  public static void main(String[] args) throws RunnerException {
+    Options opt = new OptionsBuilder()
+        .include(".*" + PulsarSenderBenchmarks.class.getSimpleName() + ".*")
+        .build();
+
+    new Runner(opt).run();
+  }
+}
+

bom/pom.xml

@@ -136,6 +136,11 @@
         <artifactId>zipkin-reporter-metrics-micrometer</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>${project.groupId}</groupId>
+        <artifactId>zipkin-sender-pulsar-client</artifactId>
+        <version>${project.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 

pom.xml

@@ -25,6 +25,7 @@
     <module>spring-beans</module>
     <module>brave</module>
     <module>metrics-micrometer</module>
+    <module>pulsar-client</module>
   </modules>
 
   <properties>

pulsar-client/bnd.bnd

@@ -0,0 +1,4 @@
+Import-Package: \
+  *
+Export-Package: \
+  zipkin2.reporter.pulsar

pulsar-client/pom.xml

@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright The OpenZipkin Authors
+    SPDX-License-Identifier: Apache-2.0
+
+-->
+<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>
+
+  <parent>
+    <artifactId>zipkin-reporter-parent</artifactId>
+    <groupId>io.zipkin.reporter2</groupId>
+    <version>3.4.4-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>zipkin-sender-pulsar-client</artifactId>
+  <name>Zipkin Sender: Pulsar Client 4.x</name>
+
+  <properties>
+    <!-- Matches Export-Package in bnd.bnd -->
+    <module.name>zipkin2.reporter.pulsar</module.name>
+
+    <main.basedir>${project.basedir}/..</main.basedir>
+    <pulsar-client.version>4.0.2</pulsar-client.version>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>zipkin-reporter</artifactId>
+      <version>${project.version}</version>
+      <!-- Senders don't use zipkin types. Excluding allows brave users to
+           avoid them by default. -->
+      <exclusions>
+        <exclusion>
+          <groupId>io.zipkin.zipkin2</groupId>
+          <artifactId>zipkin</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.pulsar</groupId>
+      <artifactId>pulsar-client</artifactId>
+      <version>${pulsar-client.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.testcontainers</groupId>
+      <artifactId>junit-jupiter</artifactId>
+      <version>${testcontainers.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <profiles>
+    <profile>
+      <id>release</id>
+      <properties>
+        <!-- pulsar 4.0+ is Java 8 bytecode -->
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <maven.compiler.release>8</maven.compiler.release>
+      </properties>
+      <build>
+        <plugins>
+          <plugin>
+            <artifactId>maven-enforcer-plugin</artifactId>
+            <version>${maven-enforcer-plugin.version}</version>
+            <executions>
+              <execution>
+                <id>enforce-java</id>
+                <goals>
+                  <goal>enforce</goal>
+                </goals>
+                <configuration>
+                  <rules>
+                    <!-- The only LTS JDK we support that can compile 8 bytecode is 11.
+                         https://www.oracle.com/java/technologies/javase/17-relnote-issues.html -->
+                    <requireJavaVersion>
+                      <version>[11,12)</version>
+                    </requireJavaVersion>
+                  </rules>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+</project>