Add single customizer API in PulsarConsumerFactory (#321)

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<T> implements PulsarConsumerFactory<T>
 		this.consumerConfig = Collections.unmodifiableMap(consumerConfig);
 	}
 
+	@Override
+	public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics,
+			@Nullable String subscriptionName, ConsumerBuilderCustomizer<T> customizer) throws PulsarClientException {
+		return createConsumer(schema, topics, subscriptionName, null,
+				customizer != null ? Collections.singletonList(customizer) : null);
+	}
+
 	@Override
 	public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics,
 			@Nullable String subscriptionName, @Nullable Map<String, String> metadataProperties,
 			@Nullable List<ConsumerBuilderCustomizer<T>> customizers) throws PulsarClientException {
+		Objects.requireNonNull(schema, "Schema must be specified");
 		ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
 		Map<String, Object> config = new HashMap<>(this.consumerConfig);
 		if (topics != null) {

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java

@@ -37,6 +37,27 @@ import org.springframework.lang.Nullable;
  */
 public interface PulsarConsumerFactory<T> {
 
+	/**
+	 * 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<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics, @Nullable String subscriptionName,
+			ConsumerBuilderCustomizer<T> customizer) throws PulsarClientException;
+
 	/**
 	 * Create a consumer.
 	 * @param schema the schema of the messages to be sent

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)) {