GH-1961: Polishing

- rename field to something more meaningful
- just compare the contexts in the event listener

**cherry-pick to 2.7.x**

Author: garyrussell

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

@@ -60,14 +60,14 @@ public class DefaultDestinationTopicResolver implements DestinationTopicResolver
 
 	private final ApplicationContext applicationContext;
 
-	private boolean containerClosed;
+	private boolean contextRefreshed;
 
 	public DefaultDestinationTopicResolver(Clock clock, ApplicationContext applicationContext) {
 		this.applicationContext = applicationContext;
 		this.clock = clock;
 		this.sourceDestinationsHolderMap = new HashMap<>();
 		this.destinationsTopicMap = new HashMap<>();
-		this.containerClosed = false;
+		this.contextRefreshed = false;
 	}
 
 	@Override
@@ -127,7 +127,7 @@ private DestinationTopic getDestinationFor(String topic) {
 	}
 
 	private DestinationTopicHolder getDestinationHolderFor(String topic) {
-		return this.containerClosed
+		return this.contextRefreshed
 				? doGetDestinationFor(topic)
 				: getDestinationTopicSynchronized(topic);
 	}
@@ -145,9 +145,9 @@ private DestinationTopicHolder doGetDestinationFor(String topic) {
 
 	@Override
 	public void addDestinationTopics(List<DestinationTopic> destinationsToAdd) {
-		if (this.containerClosed) {
+		if (this.contextRefreshed) {
 			throw new IllegalStateException("Cannot add new destinations, "
-					+ DefaultDestinationTopicResolver.class.getSimpleName() + " is already closed.");
+					+ DefaultDestinationTopicResolver.class.getSimpleName() + " is already refreshed.");
 		}
 		synchronized (this.sourceDestinationsHolderMap) {
 			this.destinationsTopicMap.putAll(destinationsToAdd
@@ -175,13 +175,18 @@ private DestinationTopic getNextDestinationTopic(List<DestinationTopic> destinat
 
 	@Override
 	public void onApplicationEvent(ContextRefreshedEvent event) {
-		if (Objects.equals(event.getApplicationContext().getId(), this.applicationContext.getId())) {
-			this.containerClosed = true;
+		if (Objects.equals(event.getApplicationContext(), this.applicationContext)) {
+			this.contextRefreshed = true;
 		}
 	}
 
-	public boolean isContainerClosed() {
-		return this.containerClosed;
+	/**
+	 * Return true if the application context is refreshed.
+	 * @return true if refreshed.
+	 * @since 2.7.8
+	 */
+	public boolean isContextRefreshed() {
+		return this.contextRefreshed;
 	}
 
 	public static class DestinationTopicHolder {

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

@@ -19,7 +19,6 @@
 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;
@@ -167,7 +166,6 @@ private long getExpectedNextExecutionTime(DestinationTopic destinationTopic) {
 
 	@Test
 	void shouldThrowIfAddsDestinationsAfterClosed() {
-		given(applicationContext.getId()).willReturn("the-context_id");
 		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
 				.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
 		assertThatIllegalStateException().isThrownBy(() ->
@@ -176,19 +174,16 @@ void shouldThrowIfAddsDestinationsAfterClosed() {
 
 	@Test
 	void shouldCloseContainerOnContextRefresh() {
-		given(applicationContext.getId()).willReturn("the-context_id");
 		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
 				.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
-		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContainerClosed()).isTrue();
+		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContextRefreshed()).isTrue();
 	}
 
 	@Test
-	void shouldNotCloseContainerOnOtherContextRefresh() {
-		given(applicationContext.getId()).willReturn("the-context_id");
-		given(otherApplicationContext.getId()).willReturn("other-context_id");
+	void shouldNotMarkContainerRefeshedOnOtherContextRefresh() {
 		((DefaultDestinationTopicResolver) defaultDestinationTopicContainer)
 				.onApplicationEvent(new ContextRefreshedEvent(otherApplicationContext));
-		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContainerClosed()).isFalse();
+		assertThat(((DefaultDestinationTopicResolver) defaultDestinationTopicContainer).isContextRefreshed()).isFalse();
 	}
 
 }