From 9b35311b303ec6f0388662fe7226da1dff1b9ae0 Mon Sep 17 00:00:00 2001
From: Chris Bono <chris.bono@gmail.com>
Date: Fri, 1 Mar 2024 22:54:43 -0600
Subject: [PATCH] Add producer interceptor cache key tests

This commit adds a test to PulsarTemplateTests that ensures producers
are properly cached when the template is configured with one or more
producer interceptors.

(cherry picked from commit cffffa6496f894baf00b54df98c9baa4c6a9f53a)
---
 .../pulsar/core/PulsarTemplateTests.java      | 31 ++++++++++++++++++-
 1 file changed, 30 insertions(+), 1 deletion(-)

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<ProducerInterceptor> interceptors) {
+	void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) throws PulsarClientException {
 		PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client, topic);
 		PulsarTemplate<String> 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<String>(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 <T> void assertCacheSize(CachingPulsarProducerFactory<T> 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 {