diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index 17a4cc69..2b4c00e3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.TreeMap; import org.apache.pulsar.client.api.Consumer; @@ -61,10 +62,18 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory this.consumerConfig = Collections.unmodifiableMap(consumerConfig); } + @Override + public Consumer createConsumer(Schema schema, @Nullable Collection topics, + @Nullable String subscriptionName, ConsumerBuilderCustomizer customizer) throws PulsarClientException { + return createConsumer(schema, topics, subscriptionName, null, + customizer != null ? Collections.singletonList(customizer) : null); + } + @Override public Consumer createConsumer(Schema schema, @Nullable Collection topics, @Nullable String subscriptionName, @Nullable Map metadataProperties, @Nullable List> customizers) throws PulsarClientException { + Objects.requireNonNull(schema, "Schema must be specified"); ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); Map config = new HashMap<>(this.consumerConfig); if (topics != null) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java index 21ad4269..c3bc1c08 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java @@ -37,6 +37,27 @@ import org.springframework.lang.Nullable; */ public interface PulsarConsumerFactory { + /** + * Create a consumer. + * @param schema the schema of the messages to be sent + * @param topics the topics the consumer will subscribe to, replacing the default + * topics, or {@code null} to use the default topics. Beware that using + * {@link ConsumerBuilder#topic} or {@link ConsumerBuilder#topics} will add to the + * default topics, not override them. Also beware that specifying {@code null} when no + * default topic is configured will result in an exception. + * @param subscriptionName the name to use for the subscription to the consumed + * topic(s) or {@code null} to use the default configured subscription name. Beware + * that specifying {@code null} when no default subscription name is configured will + * result in an exception + * @param customizer an optional customizer to apply to the consumer builder. Note + * that the customizer is applied last and has the potential for overriding any + * specified parameters or default properties. + * @return the consumer + * @throws PulsarClientException if any error occurs + */ + Consumer createConsumer(Schema schema, @Nullable Collection topics, @Nullable String subscriptionName, + ConsumerBuilderCustomizer customizer) throws PulsarClientException; + /** * Create a consumer. * @param schema the schema of the messages to be sent diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java index 45f47cb3..83016549 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java @@ -75,6 +75,13 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); } + @Test + void withoutSchema() { + assertThatThrownBy( + () -> consumerFactory.createConsumer(null, Collections.singletonList("topic0"), null, null, null)) + .isInstanceOf(NullPointerException.class).hasMessageContaining("Schema must be specified"); + } + @SuppressWarnings("resource") @Test void withSchemaOnly() { @@ -119,6 +126,14 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { } } + @Test + void withSingleCustomizerApi() throws PulsarClientException { + try (var consumer = consumerFactory.createConsumer(SCHEMA, Collections.singletonList("topic0"), + "topic0-sub", (cb) -> cb.consumerName("foo-consumer"))) { + assertThat(consumer.getConsumerName()).isEqualTo("foo-consumer"); + } + } + @Test void customizesAreAppliedLast() throws PulsarClientException { try (var consumer = consumerFactory.createConsumer(SCHEMA, Collections.singletonList("topic0"), @@ -162,6 +177,13 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, defaultConfig); } + @Test + void withoutSchema() { + assertThatThrownBy( + () -> consumerFactory.createConsumer(null, Collections.singletonList("topic0"), null, null, null)) + .isInstanceOf(NullPointerException.class).hasMessageContaining("Schema must be specified"); + } + @Test void withSchemaOnly() throws PulsarClientException { try (var consumer = consumerFactory.createConsumer(SCHEMA, null, null, null, null)) {