Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2055: Containers Must Implement DisposableBean

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/2055

If context initialization fails, `Lifecycle.stop()` is not called.
Containers must be stopped from `DisposableBean` in this case.

**cherry-pick to 2.7.x, 2.6.x, 2.5.x**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java
#	spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java
This commit is contained in:
Gary Russell
2022-01-04 13:38:38 -05:00
committed by Artem Bilan
parent 4cc23d4b23
commit 129b2bf1cb
3 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -227,14 +227,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec
@Override
public void destroy() {
for (MessageListenerContainer listenerContainer : getListenerContainers()) {
if (listenerContainer instanceof DisposableBean) {
try {
((DisposableBean) listenerContainer).destroy();
}
catch (Exception ex) {
this.logger.warn(ex, "Failed to destroy message listener container");
}
}
listenerContainer.destroy();
}
}

10
spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.lang.Nullable;
@@ -34,7 +35,7 @@ import org.springframework.lang.Nullable;
* @author Gary Russell
* @author Vladimir Tsanev
*/
public interface MessageListenerContainer extends SmartLifecycle {
public interface MessageListenerContainer extends SmartLifecycle, DisposableBean {
/**
* Setup the message listener to use. Throws an {@link IllegalArgumentException}
@@ -151,4 +152,9 @@ public interface MessageListenerContainer extends SmartLifecycle {
throw new UnsupportedOperationException("This container does not support retrieving the listener id");
}
@Override
default void destroy() {
stop();
}
}

5
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -643,8 +643,9 @@ public class KafkaMessageListenerContainerTests {
inOrder.verify(consumer).commitSync(anyMap(), any());
inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
inOrder.verify(consumer).commitSync(anyMap(), any());
container.stop();
container.destroy();
assertThat(advised).containsExactly("one", "two", "one", "two");
assertThat(container.isRunning()).isFalse();
}
@SuppressWarnings("unchecked")