From df0054299041d4bdc2a23f8a8ae9fa16db6f6637 Mon Sep 17 00:00:00 2001
From: Chris Bono <cbono@vmware.com>
Date: Sun, 15 Oct 2023 21:52:23 -0500
Subject: [PATCH] Fully clear reactive sender cache on Lifecycle stop

---
 gradle/aggregate-jacoco-report.gradle         |  2 ++
 .../DefaultReactivePulsarSenderFactory.java   | 32 +++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/gradle/aggregate-jacoco-report.gradle b/gradle/aggregate-jacoco-report.gradle
index 87c46149..48405b31 100644
--- a/gradle/aggregate-jacoco-report.gradle
+++ b/gradle/aggregate-jacoco-report.gradle
@@ -12,6 +12,8 @@ project.afterEvaluate {
 
 	tasks.create(name: 'aggregateJacocoTestReport', type: JacocoReport) {
 
+		dependsOn ':spring-pulsar-cache-provider:compileJava', ':spring-pulsar-cache-provider-caffeine:compileJava'
+
 		group = 'verification'
 		description = 'Generates aggregate code coverage report for all projects test tasks'
 
diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
index a349bbca..f98a5e9d 100644
--- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
+++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
@@ -35,6 +35,7 @@ import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * Default implementation of {@link ReactivePulsarSenderFactory}.
@@ -172,13 +173,44 @@ public final class DefaultReactivePulsarSenderFactory<T>
 	@Override
 	public void doStop() {
 		try {
+			this.reflectivelyClearCache();
 			this.reactiveMessageSenderCache.close();
+
 		}
 		catch (Exception e) {
 			throw new RuntimeException(e);
 		}
 	}
 
+	/**
+	 * Workaround to reflectively clear the underlying producer cache.
+	 *
+	 * TODO: Remove once this is supported in the Reactive client.
+	 */
+	private void reflectivelyClearCache() {
+		// reactiveMessageSenderCache
+		// (org.apache.pulsar.reactive.client.internal.adapter.ProducerCache)
+		var cacheProviderField = ReflectionUtils.findField(this.reactiveMessageSenderCache.getClass(), "cacheProvider");
+		ReflectionUtils.makeAccessible(cacheProviderField);
+
+		// org.apache.pulsar.reactive.client.producercache.CaffeineShadedProducerCacheProvider
+		var cacheProvider = ReflectionUtils.getField(cacheProviderField, this.reactiveMessageSenderCache);
+
+		// org.apache.pulsar.reactive.shade.com.github.benmanes.caffeine.cache.BoundedLocalCache$BoundedLocalAsyncCache
+		var cacheField = ReflectionUtils.findField(cacheProvider.getClass(), "cache");
+		ReflectionUtils.makeAccessible(cacheField);
+		var cache = ReflectionUtils.getField(cacheField, cacheProvider);
+
+		// org.apache.pulsar.reactive.shade.com.github.benmanes.caffeine.cache.SSLMSAW
+		var actualCacheField = ReflectionUtils.findField(cache.getClass(), "cache");
+		ReflectionUtils.makeAccessible(actualCacheField);
+		var actualCache = ReflectionUtils.getField(actualCacheField, cache);
+
+		var clearMethod = ReflectionUtils.findMethod(actualCache.getClass(), "clear");
+		ReflectionUtils.makeAccessible(clearMethod);
+		ReflectionUtils.invokeMethod(clearMethod, actualCache);
+	}
+
 	/**
 	 * Builder for {@link DefaultReactivePulsarSenderFactory}.
 	 *