GH-1961: DefaultDestinationTopicResolver: Inconsistencies In Spring Cloud Environment (#1962)

* GH-1961: Close container on ContextRefreshEvent only if event source is the same application context.

* GH-1961: Updated Javadoc

Author: yvettequinby

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolver.java

@@ -26,6 +26,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
@@ -40,6 +41,7 @@
  *
  * @author Tomaz Fernandes
  * @author Gary Russell
+ * @author Yvette Quinby
  * @since 2.7
  *
  */
@@ -56,9 +58,12 @@ public class DefaultDestinationTopicResolver implements DestinationTopicResolver
 
 	private final Clock clock;
 
+	private final ApplicationContext applicationContext;
+
 	private boolean containerClosed;
 
-	public DefaultDestinationTopicResolver(Clock clock) {
+	public DefaultDestinationTopicResolver(Clock clock, ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
 		this.clock = clock;
 		this.sourceDestinationsHolderMap = new HashMap<>();
 		this.destinationsTopicMap = new HashMap<>();
@@ -170,7 +175,13 @@ private DestinationTopic getNextDestinationTopic(List<DestinationTopic> destinat
 
 	@Override
 	public void onApplicationEvent(ContextRefreshedEvent event) {
-		this.containerClosed = true;
+		if (Objects.equals(event.getApplicationContext().getId(), this.applicationContext.getId())) {
+			this.containerClosed = true;
+		}
+	}
+
+	public boolean isContainerClosed() {
+		return this.containerClosed;
 	}
 
 	public static class DestinationTopicHolder {

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DefaultDestinationTopicResolverTests.java

@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.assertj.core.api.Assertions.assertThatNullPointerException;
+import static org.mockito.BDDMockito.given;
 
 import java.math.BigInteger;
 import java.time.Clock;
@@ -28,21 +29,34 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
 
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.kafka.listener.ListenerExecutionFailedException;
 import org.springframework.kafka.listener.TimestampedException;
 
 /**
  * @author Tomaz Fernandes
+ * @author Yvette Quinby
  * @since 2.7
  */
+@ExtendWith(MockitoExtension.class)
 class DefaultDestinationTopicResolverTests extends DestinationTopicTests {
 
 	private Map<String, DefaultDestinationTopicResolver.DestinationTopicHolder> destinationTopicMap;
 
+	@Mock
+	private ApplicationContext applicationContext;
+
+	@Mock
+	private ApplicationContext otherApplicationContext;
+
 	private final Clock clock = TestClockUtils.CLOCK;
 
-	private final DestinationTopicResolver defaultDestinationTopicContainer = new DefaultDestinationTopicResolver(clock);
+	private DestinationTopicResolver defaultDestinationTopicContainer;
 
 	private final long originalTimestamp = Instant.now(this.clock).toEpochMilli();
 
@@ -53,6 +67,7 @@ class DefaultDestinationTopicResolverTests extends DestinationTopicTests {
 	@BeforeEach
 	public void setup() {
 
+		defaultDestinationTopicContainer = new DefaultDestinationTopicResolver(clock, applicationContext);
 		defaultDestinationTopicContainer.addDestinationTopics(allFirstDestinationsTopics);
 		defaultDestinationTopicContainer.addDestinationTopics(allSecondDestinationTopics);
 		defaultDestinationTopicContainer.addDestinationTopics(allThirdDestinationTopics);
@@ -152,8 +167,28 @@ private long getExpectedNextExecutionTime(DestinationTopic destinationTopic) {
 
 	@Test
 	void shouldThrowIfAddsDestinationsAfterClosed() {
-		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).onApplicationEvent(null);
+		given(applicationContext.getId()).willReturn("the-context_id");
+		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
+				.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
 		assertThatIllegalStateException().isThrownBy(() ->
 				defaultDestinationTopicContainer.addDestinationTopics(Collections.emptyList()));
 	}
+
+	@Test
+	void shouldCloseContainerOnContextRefresh() {
+		given(applicationContext.getId()).willReturn("the-context_id");
+		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
+				.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
+		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContainerClosed()).isTrue();
+	}
+
+	@Test
+	void shouldNotCloseContainerOnOtherContextRefresh() {
+		given(applicationContext.getId()).willReturn("the-context_id");
+		given(otherApplicationContext.getId()).willReturn("other-context_id");
+		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
+				.onApplicationEvent(new ContextRefreshedEvent(otherApplicationContext));
+		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContainerClosed()).isFalse();
+	}
+
 }