Trim PulsarProducerFactory API (#319)

* The `createProducer(Schema)` is not useful
  in that most users will specify a topic and
  not rely on the default to send to

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

@@ -94,6 +94,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 	@Override
 	protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers) {
+		Objects.requireNonNull(schema, "Schema must be specified");
 		String resolveTopicName = ProducerUtils.resolveTopicName(topic, this);
 		ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName,
 				encryptionKeys == null ? null : new HashSet<>(encryptionKeys), customizers);
@@ -193,6 +194,12 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 					+ Objects.hashCode(this.customizers);
 		}
 
+		@Override
+		public String toString() {
+			return "ProducerCacheKey{" + "schema=" + this.schema + ", topic='" + this.topic + '\'' + ", encryptionKeys="
+					+ this.encryptionKeys + ", customizers=" + this.customizers + '}';
+		}
+
 	}
 
 	/**

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

@@ -21,6 +21,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 
 import org.apache.pulsar.client.api.BatcherBuilder;
 import org.apache.pulsar.client.api.CryptoKeyReader;
@@ -57,11 +58,6 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 		this.producerConfig = Collections.unmodifiableMap(config);
 	}
 
-	@Override
-	public Producer<T> createProducer(Schema<T> schema) throws PulsarClientException {
-		return doCreateProducer(schema, null, null, null);
-	}
-
 	@Override
 	public Producer<T> createProducer(Schema<T> schema, @Nullable String topic) throws PulsarClientException {
 		return doCreateProducer(schema, topic, null, null);
@@ -70,8 +66,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 	@Override
 	public Producer<T> createProducer(Schema<T> schema, @Nullable String topic,
 			@Nullable ProducerBuilderCustomizer<T> customizer) throws PulsarClientException {
-		return doCreateProducer(schema, topic, Collections.emptyList(),
-				customizer != null ? Collections.singletonList(customizer) : null);
+		return doCreateProducer(schema, topic, null, customizer != null ? Collections.singletonList(customizer) : null);
 	}
 
 	@Override
@@ -98,6 +93,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
 	protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
 			@Nullable Collection<String> encryptionKeys, @Nullable List<ProducerBuilderCustomizer<T>> customizers)
 			throws PulsarClientException {
+		Objects.requireNonNull(schema, "Schema must be specified");
 		String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
 		this.logger.trace(() -> "Creating producer for '%s' topic".formatted(resolvedTopic));
 		ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);

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

@@ -38,14 +38,6 @@ import org.springframework.lang.Nullable;
  */
 public interface PulsarProducerFactory<T> {
 
-	/**
-	 * Create a producer that will send messages to the default topic.
-	 * @param schema the schema of the messages to be sent
-	 * @return the producer
-	 * @throws PulsarClientException if any error occurs
-	 */
-	Producer<T> createProducer(Schema<T> schema) throws PulsarClientException;
-
 	/**
 	 * Create a producer.
 	 * @param schema the schema of the messages to be sent
@@ -61,7 +53,7 @@ public interface PulsarProducerFactory<T> {
 	 * @param schema the schema of the messages to be sent
 	 * @param topic the topic the producer will send messages to or {@code null} to use
 	 * the default topic
-	 * @param customizer optional producer builder customizer
+	 * @param customizer the optional customizer to apply to the producer builder
 	 * @return the producer
 	 * @throws PulsarClientException if any error occurs
 	 */

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java

@@ -34,8 +34,7 @@ import org.junit.jupiter.api.Test;
 class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	@Test
-	@SuppressWarnings("unchecked")
-	void createProducerMultipleCalls() throws PulsarClientException {
+	void createProducerMultipleTimeDoesNotCacheProducer() throws PulsarClientException {
 		Map<String, Object> producerConfig = Collections.emptyMap();
 		PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, producerConfig);
 		try (Producer<String> producer1 = producerFactory.createProducer(schema, "topic1")) {

spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java

@@ -18,6 +18,7 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatNullPointerException;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
@@ -62,6 +63,7 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 		pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
 	}
 
+	@SuppressWarnings("ConstantConditions")
 	@AfterEach
 	void closePulsarClient() throws PulsarClientException {
 		if (pulsarClient != null && !pulsarClient.isClosed()) {
@@ -69,20 +71,6 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 		}
 	}
 
-	@Test
-	@SuppressWarnings("unchecked")
-	void createProducerWithAllOptions() throws PulsarClientException {
-		var keys = Set.of("key");
-		ProducerBuilderCustomizer<String> customizer1 = mock(ProducerBuilderCustomizer.class);
-		var producerFactory = newProducerFactory();
-		try (var producer = producerFactory.createProducer(schema, "topic0", keys,
-				Collections.singletonList(customizer1))) {
-			assertThatProducerHasSchemaAndTopic(producer, schema, "topic0");
-			assertThatProducerHasEncryptionKeys(producer, keys);
-			verify(customizer1).customize(any(ProducerBuilder.class));
-		}
-	}
-
 	private void assertThatProducerHasSchemaAndTopic(Producer<String> producer, Schema<String> expectedSchema,
 			String expectedTopic) {
 		producer = actualProducer(producer);
@@ -132,34 +120,27 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 	protected abstract PulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
 			Map<String, Object> producerConfig);
 
-	@Nested
-	class CreateProducerSchemaOnlyApi {
-
-		@Test
-		void withDefaultTopic() throws PulsarClientException {
-			var producerFactory = newProducerFactoryWithDefaultTopic("topic0");
-			try (var producer = producerFactory.createProducer(schema)) {
-				assertThatProducerHasSchemaAndTopic(producer, schema, "topic0");
-			}
+	@Test
+	@SuppressWarnings("unchecked")
+	void createProducerWithAllOptions() throws PulsarClientException {
+		var keys = Set.of("key");
+		ProducerBuilderCustomizer<String> customizer1 = mock(ProducerBuilderCustomizer.class);
+		var producerFactory = newProducerFactory();
+		try (var producer = producerFactory.createProducer(schema, "topic0", keys,
+				Collections.singletonList(customizer1))) {
+			assertThatProducerHasSchemaAndTopic(producer, schema, "topic0");
+			assertThatProducerHasEncryptionKeys(producer, keys);
+			verify(customizer1).customize(any(ProducerBuilder.class));
 		}
-
-		@Test
-		void withoutDefaultTopic() {
-			assertThatIllegalArgumentException().isThrownBy(() -> newProducerFactory().createProducer(schema))
-					.withMessageContaining("Topic must be specified when no default topic is configured");
-		}
-
 	}
 
 	@Nested
 	class CreateProducerSchemaAndTopicApi {
 
 		@Test
-		void topicSpecifiedWithDefaultTopic() throws PulsarClientException {
-			var producerFactory = newProducerFactoryWithDefaultTopic("topic0");
-			try (var producer = producerFactory.createProducer(schema, "topic1")) {
-				assertThatProducerHasSchemaAndTopic(producer, schema, "topic1");
-			}
+		void withoutSchema() {
+			assertThatNullPointerException().isThrownBy(() -> newProducerFactory().createProducer(null, "topic0"))
+					.withMessageContaining("Schema must be specified");
 		}
 
 		@Test
@@ -170,10 +151,10 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 		}
 
 		@Test
-		void noTopicSpecifiedWithDefaultTopic() throws PulsarClientException {
+		void topicSpecifiedWithDefaultTopic() throws PulsarClientException {
 			var producerFactory = newProducerFactoryWithDefaultTopic("topic0");
-			try (var producer = producerFactory.createProducer(schema, null)) {
-				assertThatProducerHasTopic(producer, "topic0");
+			try (var producer = producerFactory.createProducer(schema, "topic1")) {
+				assertThatProducerHasSchemaAndTopic(producer, schema, "topic1");
 			}
 		}
 
@@ -183,6 +164,14 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport
 					.withMessageContaining("Topic must be specified when no default topic is configured");
 		}
 
+		@Test
+		void noTopicSpecifiedWithDefaultTopic() throws PulsarClientException {
+			var producerFactory = newProducerFactoryWithDefaultTopic("topic0");
+			try (var producer = producerFactory.createProducer(schema, null)) {
+				assertThatProducerHasTopic(producer, "topic0");
+			}
+		}
+
 	}
 
 	@Nested