SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Close ProducerFactory on unbind

Browse Source
This commit is contained in:
Marius Bogoevici
2016-08-23 16:12:24 -04:00
parent 1c353da2e1
commit 73460aa12e
1 changed files with 9 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -63,7 +63,6 @@ import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.listener.config.ContainerProperties;
@@ -206,12 +205,12 @@ public class KafkaMessageChannelBinder extends
this.topicsInUse.put(destination, partitions); this.topicsInUse.put(destination, partitions);
ProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties); DefaultKafkaProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFB); KafkaTemplate<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFB);
if (this.producerListener != null) { if (this.producerListener != null) {
kafkaTemplate.setProducerListener(this.producerListener); kafkaTemplate.setProducerListener(this.producerListener);
} }
return new ProducerConfigurationMessageHandler(kafkaTemplate, destination, producerProperties); return new ProducerConfigurationMessageHandler(kafkaTemplate, destination, producerProperties, producerFB);
} }
@Override @Override
@@ -233,7 +232,7 @@ public class KafkaMessageChannelBinder extends
return name; return name;
} }
private ProducerFactory<byte[], byte[]> getProducerFactory( private DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) { ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
Map<String, Object> props = new HashMap<>(); Map<String, Object> props = new HashMap<>();
if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) { if (!ObjectUtils.isEmpty(configurationProperties.getConfiguration())) {
@@ -550,8 +549,11 @@ public class KafkaMessageChannelBinder extends
private boolean running = true; private boolean running = true;
private final DefaultKafkaProducerFactory<byte[], byte[]> producerFactory;
private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic, private ProducerConfigurationMessageHandler(KafkaTemplate<byte[], byte[]> kafkaTemplate, String topic,
ExtendedProducerProperties<KafkaProducerProperties> producerProperties) { ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
DefaultKafkaProducerFactory<byte[], byte[]> producerFactory) {
super(kafkaTemplate); super(kafkaTemplate);
setTopicExpression(new LiteralExpression(topic)); setTopicExpression(new LiteralExpression(topic));
setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory()); setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
@@ -562,6 +564,7 @@ public class KafkaMessageChannelBinder extends
if (producerProperties.getExtension().isSync()) { if (producerProperties.getExtension().isSync()) {
setSync(true); setSync(true);
} }
this.producerFactory = producerFactory;
} }
@Override @Override
@@ -577,6 +580,7 @@ public class KafkaMessageChannelBinder extends
@Override @Override
public void stop() { public void stop() {
producerFactory.stop();
this.running = false; this.running = false;
} }