io.zipkin.brave
diff --git a/zipkin-server/src/main/java/zipkin2/server/internal/InternalZipkinConfiguration.java b/zipkin-server/src/main/java/zipkin2/server/internal/InternalZipkinConfiguration.java
index 32a03838d2..db63feabbf 100644
--- a/zipkin-server/src/main/java/zipkin2/server/internal/InternalZipkinConfiguration.java
+++ b/zipkin-server/src/main/java/zipkin2/server/internal/InternalZipkinConfiguration.java
@@ -16,6 +16,7 @@
import zipkin2.server.internal.mysql.ZipkinMySQLStorageConfiguration;
import zipkin2.server.internal.prometheus.ZipkinMetricsController;
import zipkin2.server.internal.prometheus.ZipkinPrometheusMetricsConfiguration;
+import zipkin2.server.internal.pulsar.ZipkinPulsarCollectorConfiguration;
import zipkin2.server.internal.rabbitmq.ZipkinRabbitMQCollectorConfiguration;
import zipkin2.server.internal.scribe.ZipkinScribeCollectorConfiguration;
import zipkin2.server.internal.ui.ZipkinUiConfiguration;
@@ -35,7 +36,8 @@
ZipkinGrpcCollector.class,
ZipkinActiveMQCollectorConfiguration.class,
ZipkinKafkaCollectorConfiguration.class,
- ZipkinRabbitMQCollectorConfiguration.class,
+ ZipkinRabbitMQCollectorConfiguration.class,
+ ZipkinPulsarCollectorConfiguration.class,
ZipkinMetricsController.class,
ZipkinHealthController.class,
ZipkinPrometheusMetricsConfiguration.class,
diff --git a/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfiguration.java b/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfiguration.java
new file mode 100644
index 0000000000..af06a2d8b3
--- /dev/null
+++ b/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfiguration.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.server.internal.pulsar;
+
+import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import zipkin2.collector.CollectorMetrics;
+import zipkin2.collector.CollectorSampler;
+import zipkin2.collector.pulsar.PulsarCollector;
+import zipkin2.storage.StorageComponent;
+
+import static io.micrometer.common.util.StringUtils.isEmpty;
+
+/** Auto-configuration for {@link PulsarCollector}. */
+@ConditionalOnClass(PulsarCollector.class)
+@Conditional(ZipkinPulsarCollectorConfiguration.PulsarConditions.class)
+@EnableConfigurationProperties(ZipkinPulsarCollectorProperties.class)
+public class ZipkinPulsarCollectorConfiguration {
+
+ @Bean(initMethod = "start")
+ PulsarCollector pulsar(
+ ZipkinPulsarCollectorProperties properties,
+ CollectorSampler sampler,
+ CollectorMetrics metrics,
+ StorageComponent storage
+ ) {
+ return properties.toBuilder().sampler(sampler).metrics(metrics).storage(storage).build();
+ }
+
+ /**
+ * This condition passes when {@link ZipkinPulsarCollectorProperties#getServiceUrl()} is set
+ * to non-empty.
+ *
+ * This is here because the yaml defaults this property to empty like this, and spring-boot
+ * doesn't have an option to treat empty properties as unset.
+ *
+ *
{@code
+ * service-url: ${PULSAR_SERVICE_URL:}
+ * }
+ */
+ static final class PulsarConditions extends AllNestedConditions {
+
+ PulsarConditions() {
+ super(ConfigurationPhase.REGISTER_BEAN);
+ }
+
+ @ConditionalOnProperty(prefix = "zipkin.collector.pulsar", name = "enabled",
+ havingValue = "true", matchIfMissing = true)
+ private static final class PulsarEnabledCondition {
+ }
+
+ @Conditional(PulsarServiceUrlCondition.class)
+ private static final class PulsarServiceUrlCondition implements Condition {
+ @Override public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+ String serviceUrl = context.getEnvironment().getProperty("zipkin.collector.pulsar.service-url");
+ return !isEmpty(serviceUrl);
+ }
+ }
+ }
+}
diff --git a/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorProperties.java b/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorProperties.java
new file mode 100644
index 0000000000..28f984c41a
--- /dev/null
+++ b/zipkin-server/src/main/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorProperties.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.server.internal.pulsar;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import zipkin2.collector.pulsar.PulsarCollector;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/** Properties for configuring and building a {@link PulsarCollector}. */
+@ConfigurationProperties("zipkin.collector.pulsar")
+class ZipkinPulsarCollectorProperties {
+
+ /** The service URL for the Pulsar service. */
+ private String serviceUrl;
+ /** Pulsar topic span data will be retrieved from. */
+ private String topic;
+ /** Specify the subscription name for this consumer. */
+ private String subscriptionName;
+ /** Number of concurrent span consumers */
+ private Integer concurrency;
+ /** Additional Pulsar client configuration. */
+ private Map clientProps = new LinkedHashMap<>();
+ /** Additional Pulsar consumer configuration. */
+ private Map consumerProps = new LinkedHashMap<>();
+
+ public String getServiceUrl() {
+ return serviceUrl;
+ }
+
+ public void setServiceUrl(String serviceUrl) {
+ this.serviceUrl = serviceUrl;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getSubscriptionName() {
+ return subscriptionName;
+ }
+
+ public void setSubscriptionName(String subscriptionName) {
+ this.subscriptionName = subscriptionName;
+ }
+
+ public Integer getConcurrency() {
+ return concurrency;
+ }
+
+ public void setConcurrency(Integer concurrency) {
+ this.concurrency = concurrency;
+ }
+
+ public Map getClientProps() {
+ return clientProps;
+ }
+
+ public void setClientProps(Map clientProps) {
+ this.clientProps = clientProps;
+ }
+
+ public Map getConsumerProps() {
+ return consumerProps;
+ }
+
+ public void setConsumerProps(Map consumerProps) {
+ this.consumerProps = consumerProps;
+ }
+
+ public PulsarCollector.Builder toBuilder() {
+ final PulsarCollector.Builder result = PulsarCollector.builder();
+ if (serviceUrl != null) {
+ result.serviceUrl(serviceUrl);
+ }
+ if (topic != null) result.topic(topic);
+ if (concurrency != null) result.concurrency(concurrency);
+ if (subscriptionName != null) result.subscriptionName(subscriptionName);
+ if (!clientProps.isEmpty()) result.clientProps(clientProps);
+ if (!consumerProps.isEmpty()) result.consumerProps(consumerProps);
+ return result;
+ }
+}
diff --git a/zipkin-server/src/main/resources/zipkin-server-shared.yml b/zipkin-server/src/main/resources/zipkin-server-shared.yml
index f8f1b128a7..4dcf40d2ac 100644
--- a/zipkin-server/src/main/resources/zipkin-server-shared.yml
+++ b/zipkin-server/src/main/resources/zipkin-server-shared.yml
@@ -59,6 +59,16 @@ zipkin:
enabled: ${COLLECTOR_SCRIBE_ENABLED:${SCRIBE_ENABLED:false}}
category: ${SCRIBE_CATEGORY:zipkin}
port: ${COLLECTOR_PORT:9410}
+ pulsar:
+ enabled: ${COLLECTOR_PULSAR_ENABLED:true}
+ # The service URL for the Pulsar client ex. pulsar://my-broker:6650.
+ service-url: ${PULSAR_SERVICE_URL:}
+ # Queue zipkin spans will be consumed from.
+ topic: ${PULSAR_TOPIC:zipkin}
+ # Specify the subscription name for this consumer.
+ subscription-name: ${PULSAR_SUBSCRIPTION_NAME:zipkin}
+ # Number of concurrent span consumers.
+ concurrency: ${PULSAR_CONCURRENCY:1}
discovery:
eureka:
diff --git a/zipkin-server/src/test/java/zipkin2/collector/pulsar/ZipkinPulsarCollectorPropertiesOverrideTest.java b/zipkin-server/src/test/java/zipkin2/collector/pulsar/ZipkinPulsarCollectorPropertiesOverrideTest.java
new file mode 100644
index 0000000000..38067c21f3
--- /dev/null
+++ b/zipkin-server/src/test/java/zipkin2/collector/pulsar/ZipkinPulsarCollectorPropertiesOverrideTest.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.collector.pulsar;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import zipkin2.server.internal.pulsar.Access;
+
+import java.util.List;
+import java.util.function.Function;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ZipkinPulsarCollectorPropertiesOverrideTest {
+
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+
+ @AfterEach void close() {
+ if (context != null) context.close();
+ }
+
+ public String property;
+ public Object value;
+ public Function builderExtractor;
+
+ public static Iterable data() {
+ return List.of(
+ // intentionally punting on comma-separated form of a list of addresses as it doesn't fit
+ // this unit test. Better to make a separate one than force-fit!
+ parameters("service-url", "pulsar://127.0.0.1:6650", b -> b.clientProps.get("serviceUrl")),
+ parameters("topic", "zipkin", b -> b.topic),
+ parameters("concurrency", 2, b -> b.concurrency),
+ parameters("clientProps.serviceUrl", "pulsar://127.0.0.1:6650", b -> b.clientProps.get("serviceUrl")),
+ parameters("consumerProps.subscriptionName", "zipkin-subscription", b -> b.consumerProps.get("subscriptionName"))
+ );
+ }
+
+ /** to allow us to define with a lambda */
+ static Object[] parameters(
+ String propertySuffix, T value, Function builderExtractor) {
+ return new Object[]{"zipkin.collector.pulsar." + propertySuffix, value, builderExtractor};
+ }
+
+ @MethodSource("data")
+ @ParameterizedTest(name = "{0}")
+ void propertyTransferredToCollectorBuilder(String property, Object value,
+ Function builderExtractor) throws Exception {
+ initZipkinPulsarCollectorPropertiesOverrideTest(property, value, builderExtractor);
+ TestPropertyValues.of(property + ":" + value).applyTo(context);
+ Access.registerPulsarProperties(context);
+ context.refresh();
+
+ assertThat(Access.collectorBuilder(context))
+ .extracting(builderExtractor)
+ .isEqualTo(value);
+ }
+
+ void initZipkinPulsarCollectorPropertiesOverrideTest(String property, Object value,
+ Function builderExtractor) {
+ this.property = property;
+ this.value = value;
+ this.builderExtractor = builderExtractor;
+ }
+}
diff --git a/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/Access.java b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/Access.java
new file mode 100644
index 0000000000..6f9c111ab7
--- /dev/null
+++ b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/Access.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.server.internal.pulsar;
+
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import zipkin2.collector.pulsar.PulsarCollector;
+
+/** opens package access for testing */
+public final class Access {
+
+ /** Just registering properties to avoid automatically connecting to a Pulsar server */
+ public static void registerPulsarProperties(AnnotationConfigApplicationContext context) {
+ context.register(
+ PropertyPlaceholderAutoConfiguration.class, EnablePulsarCollectorProperties.class);
+ }
+
+ @Configuration
+ @EnableConfigurationProperties(ZipkinPulsarCollectorProperties.class)
+ static class EnablePulsarCollectorProperties {
+ }
+
+ public static PulsarCollector.Builder collectorBuilder(
+ AnnotationConfigApplicationContext context) {
+ return context.getBean(ZipkinPulsarCollectorProperties.class).toBuilder();
+ }
+}
diff --git a/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfigurationTest.java b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfigurationTest.java
new file mode 100644
index 0000000000..60f42a684e
--- /dev/null
+++ b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorConfigurationTest.java
@@ -0,0 +1,82 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.server.internal.pulsar;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import zipkin2.collector.pulsar.PulsarCollector;
+import zipkin2.server.internal.InMemoryConfiguration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType;
+
+class ZipkinPulsarCollectorConfigurationTest {
+
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+
+ @AfterEach void close() {
+ context.close();
+ }
+
+ @Test void doesNotProvideCollectorComponent_whenBootstrapServersUnset() {
+ assertThatExceptionOfType(NoSuchBeanDefinitionException.class).isThrownBy(() -> {
+ context.register(
+ PropertyPlaceholderAutoConfiguration.class,
+ ZipkinPulsarCollectorConfiguration.class,
+ InMemoryConfiguration.class);
+ context.refresh();
+ context.getBean(PulsarCollector.class);
+ });
+ }
+
+ @Test void providesCollectorComponent_whenServiceUrlEmptyString() {
+ assertThatExceptionOfType(NoSuchBeanDefinitionException.class).isThrownBy(() -> {
+ TestPropertyValues.of("zipkin.collector.pulsar.service-url:").applyTo(context);
+ context.register(
+ PropertyPlaceholderAutoConfiguration.class,
+ ZipkinPulsarCollectorConfiguration.class,
+ InMemoryConfiguration.class);
+ context.refresh();
+ context.getBean(PulsarCollector.class);
+ });
+ }
+
+ @Test void providesCollectorComponent_whenServiceUrlServersSet() {
+ TestPropertyValues.of("zipkin.collector.pulsar.service-url:pulsar://localhost:6650",
+ "zipkin.collector.pulsar.subscription-name:zipkin-subscriptionName")
+ .applyTo(context);
+ context.register(
+ PropertyPlaceholderAutoConfiguration.class,
+ ZipkinPulsarCollectorConfiguration.class,
+ InMemoryConfiguration.class);
+ try {
+ context.refresh();
+ failBecauseExceptionWasNotThrown(BeanCreationException.class);
+ } catch (BeanCreationException e) {
+ assertThat(e.getCause()).hasMessageContaining(
+ "Pulsar Client is unable to subscribe to the topic");
+ }
+ }
+
+ @Test void doesNotProvidesCollectorComponent_whenServiceUrlSetAndDisabled() {
+ assertThatExceptionOfType(NoSuchBeanDefinitionException.class).isThrownBy(() -> {
+ TestPropertyValues.of("zipkin.collector.pulsar.service-url:pulsar://127.0.0.1:6650")
+ .applyTo(context);
+ TestPropertyValues.of("zipkin.collector.pulsar.enabled:false").applyTo(context);
+ context.register(
+ PropertyPlaceholderAutoConfiguration.class,
+ ZipkinPulsarCollectorConfiguration.class,
+ InMemoryConfiguration.class);
+ context.refresh();
+ context.getBean(PulsarCollector.class);
+ });
+ }
+}
diff --git a/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorPropertiesTest.java b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorPropertiesTest.java
new file mode 100644
index 0000000000..a0e2192c06
--- /dev/null
+++ b/zipkin-server/src/test/java/zipkin2/server/internal/pulsar/ZipkinPulsarCollectorPropertiesTest.java
@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenZipkin Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package zipkin2.server.internal.pulsar;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ZipkinPulsarCollectorPropertiesTest {
+ @Test void stringPropertiesConvertEmptyStringsToNull() {
+ final ZipkinPulsarCollectorProperties properties = new ZipkinPulsarCollectorProperties();
+ properties.setServiceUrl("pulsar://127.0.0.1:6650");
+ properties.setTopic("zipkin");
+ properties.setClientProps(new HashMap<>());
+ properties.setConsumerProps(new HashMap<>());
+ assertThat(properties.getServiceUrl()).isEqualTo("pulsar://127.0.0.1:6650");
+ assertThat(properties.getTopic()).isEqualTo("zipkin");
+ assertThat(properties.getClientProps()).isEmpty();
+ assertThat(properties.getConsumerProps()).isEmpty();
+ }
+}