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

Destroy internal TaskScheduler in container

Browse Source 
**Cherry-pick to 2.0.x & 1.3.x**
This commit is contained in:
Yang Qiju
2017-11-24 01:32:10 +08:00
committed by Artem Bilan
parent 11fd8637a2
commit 8c65d62a24
1 changed files with 8 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -87,6 +87,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
* @author Artem Bilan
* @author Loic Talhouarne
* @author Vladimir Tsanev
* @author Yang Qiju
*/
public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
@@ -343,6 +344,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private boolean fatalError;
private boolean taskSchedulerExplicitlySet;
@SuppressWarnings("unchecked")
ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
Assert.state(!this.isAnyManualAck || !this.autoCommit,
@@ -406,13 +409,14 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
if (this.transactionManager != null) {
this.transactionTemplate = new TransactionTemplate(this.transactionManager);
Assert.state(!(this.errorHandler instanceof RemainingRecordsErrorHandler),
"You cannot use a 'RemainingRecordsErrorHandler' with transactions");
"You cannot use a 'RemainingRecordsErrorHandler' with transactions");
}
else {
this.transactionTemplate = null;
}
if (this.containerProperties.getScheduler() != null) {
this.taskScheduler = this.containerProperties.getScheduler();
this.taskSchedulerExplicitlySet = true;
}
else {
ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
@@ -663,6 +667,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
KafkaMessageListenerContainer.this.stop();
}
this.monitorTask.cancel(true);
if (!this.taskSchedulerExplicitlySet) {
((ThreadPoolTaskScheduler) this.taskScheduler).destroy();
}
this.consumer.close();
if (this.logger.isInfoEnabled()) {
this.logger.info("Consumer stopped");