diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java index 6d373178..1f64083c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -94,6 +94,7 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor @Override protected Producer doCreateProducer(Schema schema, @Nullable String topic, @Nullable Collection encryptionKeys, @Nullable List> customizers) { + Objects.requireNonNull(schema, "Schema must be specified"); String resolveTopicName = ProducerUtils.resolveTopicName(topic, this); ProducerCacheKey producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName, encryptionKeys == null ? null : new HashSet<>(encryptionKeys), customizers); @@ -193,6 +194,12 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor + Objects.hashCode(this.customizers); } + @Override + public String toString() { + return "ProducerCacheKey{" + "schema=" + this.schema + ", topic='" + this.topic + '\'' + ", encryptionKeys=" + + this.encryptionKeys + ", customizers=" + this.customizers + '}'; + } + } /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index 06050171..250618f8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/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 implements PulsarProducerFactory this.producerConfig = Collections.unmodifiableMap(config); } - @Override - public Producer createProducer(Schema schema) throws PulsarClientException { - return doCreateProducer(schema, null, null, null); - } - @Override public Producer createProducer(Schema schema, @Nullable String topic) throws PulsarClientException { return doCreateProducer(schema, topic, null, null); @@ -70,8 +66,7 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public Producer createProducer(Schema schema, @Nullable String topic, @Nullable ProducerBuilderCustomizer 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 implements PulsarProducerFactory protected Producer doCreateProducer(Schema schema, @Nullable String topic, @Nullable Collection encryptionKeys, @Nullable List> 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 producerBuilder = this.pulsarClient.newProducer(schema); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index eeaf2673..984c74d0 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -38,14 +38,6 @@ import org.springframework.lang.Nullable; */ public interface PulsarProducerFactory { - /** - * 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 createProducer(Schema schema) throws PulsarClientException; - /** * Create a producer. * @param schema the schema of the messages to be sent @@ -61,7 +53,7 @@ public interface PulsarProducerFactory { * @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 */ diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java index e8ba5078..89cebbac 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java +++ b/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 producerConfig = Collections.emptyMap(); PulsarProducerFactory producerFactory = producerFactory(pulsarClient, producerConfig); try (Producer producer1 = producerFactory.createProducer(schema, "topic1")) { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java index 29fff250..884c38ba 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java +++ b/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 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 producer, Schema expectedSchema, String expectedTopic) { producer = actualProducer(producer); @@ -132,34 +120,27 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport protected abstract PulsarProducerFactory producerFactory(PulsarClient pulsarClient, Map 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 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