GH-3375: Override close() in ExtendedKafkaConsumer

Fixes: #3375 

* ExtendedKafkaConsumer now overrides `close()` to make sure `listener.consumerRemoved()` is called when the consumer is closed.

**Auto-cherry-pick to `3.2.x`**

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -72,6 +72,7 @@ import org.springframework.util.StringUtils;
  * @author Artem Bilan
  * @author Chris Gilbert
  * @author Adrian Gygax
+ * @author Yaniv Nahoum
  */
 public class DefaultKafkaConsumerFactory<K, V> extends KafkaResourceFactory
 		implements ConsumerFactory<K, V>, BeanNameAware, ApplicationContextAware {
@@ -519,10 +520,19 @@ public class DefaultKafkaConsumerFactory<K, V> extends KafkaResourceFactory
 			}
 		}
 
+		@Override
+		public void close() {
+			super.close();
+			notifyConsumerRemoved();
+		}
+
 		@Override
 		public void close(Duration timeout) {
 			super.close(timeout);
+			notifyConsumerRemoved();
+		}
 
+		private void notifyConsumerRemoved() {
 			for (Listener<K, V> listener : DefaultKafkaConsumerFactory.this.listeners) {
 				listener.consumerRemoved(this.idForListeners, this);
 			}

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java

@@ -42,6 +42,8 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -66,6 +68,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  * @author Artem Bilan
  * @author Adrian Gygax
  * @author Soby Chacko
+ * @author Yaniv Nahoum
  *
  * @since 1.0.6
  */
@@ -458,8 +461,9 @@ public class DefaultKafkaConsumerFactoryTests {
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	@Test
-	void listener() {
+	@ParameterizedTest
+	@ValueSource(booleans = { true, false })
+	void listener(boolean closeWithTimeout) {
 		Map<String, Object> consumerConfig = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka);
 		consumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, "foo-0");
 		DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(consumerConfig);
@@ -484,8 +488,13 @@ public class DefaultKafkaConsumerFactoryTests {
 		Consumer consumer = cf.createConsumer();
 		assertThat(adds).hasSize(1);
 		assertThat(adds.get(0)).isEqualTo("cf.foo-0");
-		assertThat(removals).hasSize(0);
-		consumer.close(Duration.ofSeconds(10));
+		assertThat(removals).isEmpty();
+		if (closeWithTimeout) {
+			consumer.close(Duration.ofSeconds(10));
+		}
+		else {
+			consumer.close();
+		}
 		assertThat(removals).hasSize(1);
 	}