diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index e27227ee..636671a9 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -198,7 +198,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @ParameterizedTest(name = "{0}") @MethodSource("interceptorInvocationTestProvider") - void interceptorInvocationTest(String topic, List interceptors) { + void interceptorInvocationTest(String topic, List interceptors) throws PulsarClientException { PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, topic); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); pulsarTemplate.send("test-interceptor"); @@ -215,6 +215,35 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class)))); } + @Test + void interceptorUsedAsCacheKeyProperly() throws PulsarClientException { + var producerFactory = new CachingPulsarProducerFactory(client, null, null, new DefaultTopicResolver(), + Duration.ofSeconds(10L), 10L, 10); + try { + var interceptors = List.of(mock(ProducerInterceptor.class)); + var pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); + assertCacheSize(producerFactory, 0); + for (int i = 0; i < 3; i++) { + pulsarTemplate.send("test-intercept-topic", "test-interceptor-" + i); + assertCacheSize(producerFactory, 1); + } + assertCacheSize(producerFactory, 1); + } + finally { + // The CPPF returns producers that do not actually close when the template + // calls close on them - destroy does close the producers though + if (producerFactory != null) { + producerFactory.destroy(); + } + } + } + + private void assertCacheSize(CachingPulsarProducerFactory producerFactory, int expectedSize) { + assertThat(producerFactory).extracting("producerCache.cache.cache") + .asInstanceOf(InstanceOfAssertFactories.MAP) + .hasSize(expectedSize); + } + @ParameterizedTest @ValueSource(booleans = { true, false }) void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception {