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

Cleaning up Kafka producer factories

Browse Source 
Calling DisposableBean.destroy() on manually created
Kafka producer factories. This affects both Kafka and
Kafka Streams binders (in the case of Kafka Streams binder,
it only matters when using the DLQ support).

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1034
This commit is contained in:
Soby Chacko
2021-03-02 16:24:34 -05:00
committed by Gary Russell
parent bc02da2900
commit e2eca34e4b
8 changed files with 52 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2021 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.
@@ -81,7 +81,8 @@ public class GlobalKTableBinder extends
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group,
getApplicationContext(), this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(),
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget));
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget),
this.kafkaStreamsBindingInformationCatalogue);
return new DefaultBinding<>(name, group, inputTarget, null);
}

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2020 the original author or authors.
* Copyright 2017-2021 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.
@@ -105,7 +105,9 @@ class KStreamBinder extends
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group,
getApplicationContext(), this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(), this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget));
this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(),
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget),
this.kafkaStreamsBindingInformationCatalogue);
return new DefaultBinding<>(name, group, inputTarget, null);
}

6
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2021 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.
@@ -83,7 +83,9 @@ class KTableBinder extends
final RetryTemplate retryTemplate = buildRetryTemplate(properties);
KafkaStreamsBinderUtils.prepareConsumerBinding(name, group,
getApplicationContext(), this.kafkaTopicProvisioner,
this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(), this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget));
this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(),
this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget),
this.kafkaStreamsBindingInformationCatalogue);
return new DefaultBinding<>(name, group, inputTarget, null);
}

7
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2021 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.
@@ -73,7 +73,8 @@ final class KafkaStreamsBinderUtils {
KafkaStreamsBinderConfigurationProperties binderConfigurationProperties,
ExtendedConsumerProperties<KafkaStreamsConsumerProperties> properties,
RetryTemplate retryTemplate,
ConfigurableListableBeanFactory beanFactory, String bindingName) {
ConfigurableListableBeanFactory beanFactory, String bindingName,
KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) {
ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties =
(ExtendedConsumerProperties) properties;
@@ -109,6 +110,8 @@ final class KafkaStreamsBinderUtils {
new ExtendedProducerProperties<>(
extendedConsumerProperties.getExtension().getDlqProducerProperties()),
binderConfigurationProperties);
kafkaStreamsBindingInformationCatalogue.addDlqProducerFactory(producerFactory);
KafkaOperations<byte[], byte[]> kafkaTemplate = new KafkaTemplate<>(producerFactory);
Map<String, DlqDestinationResolver> dlqDestinationResolvers =

17
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2021 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.
@@ -16,8 +16,10 @@
package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -31,6 +33,7 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.core.ResolvableType;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.core.ProducerFactory;
/**
* A catalogue that provides binding information for Kafka Streams target types such as
@@ -41,7 +44,7 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean;
*
* @author Soby Chacko
*/
class KafkaStreamsBindingInformationCatalogue {
public class KafkaStreamsBindingInformationCatalogue {
private final Map<KStream<?, ?>, BindingProperties> bindingProperties = new ConcurrentHashMap<>();
@@ -55,6 +58,8 @@ class KafkaStreamsBindingInformationCatalogue {
private final Map<Object, String> bindingNamesPerTarget = new HashMap<>();
private final List<ProducerFactory<byte[], byte[]>> dlqProducerFactories = new ArrayList<>();
/**
* For a given bounded {@link KStream}, retrieve it's corresponding destination on the
* broker.
@@ -175,4 +180,12 @@ class KafkaStreamsBindingInformationCatalogue {
String bindingNamePerTarget(Object target) {
return this.bindingNamesPerTarget.get(target);
}
public List<ProducerFactory<byte[], byte[]>> getDlqProducerFactories() {
return this.dlqProducerFactories;
}
public void addDlqProducerFactory(ProducerFactory<byte[], byte[]> producerFactory) {
this.dlqProducerFactories.add(producerFactory);
}
}

7
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2021 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.
@@ -18,9 +18,11 @@ package org.springframework.cloud.stream.binder.kafka.streams;
import java.util.Set;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.core.ProducerFactory;
/**
* Iterate through all {@link StreamsBuilderFactoryBean} in the application context and
@@ -108,6 +110,9 @@ class StreamsBuilderFactoryManager implements SmartLifecycle {
this.listener.streamsRemoved("streams." + n++, streamsBuilderFactoryBean.getKafkaStreams());
}
}
for (ProducerFactory<byte[], byte[]> dlqProducerFactory : this.kafkaStreamsBindingInformationCatalogue.getDlqProducerFactories()) {
((DisposableBean) dlqProducerFactory).destroy();
}
}
catch (Exception ex) {
throw new IllegalStateException(ex);

8
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2021 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.
@@ -36,6 +36,7 @@ import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBindingInformationCatalogue;
import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver;
import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction;
import org.springframework.context.ConfigurableApplicationContext;
@@ -67,7 +68,7 @@ public class DlqDestinationResolverTests {
SpringApplication app = new SpringApplication(WordCountProcessorApplication.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
try (ConfigurableApplicationContext context = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.function.definition=process",
@@ -104,6 +105,9 @@ public class DlqDestinationResolverTests {
ConsumerRecord<String, String> cr2 = KafkaTestUtils.getSingleRecord(consumer1,
"topic2-dlq");
assertThat(cr2.value()).isEqualTo("foobar");
final KafkaStreamsBindingInformationCatalogue catalogue = context.getBean(KafkaStreamsBindingInformationCatalogue.class);
assertThat(catalogue.getDlqProducerFactories().size()).isEqualTo(1);
}
finally {
pf.destroy();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2021 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.
@@ -79,7 +79,6 @@ import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.config.MessageSourceCustomizer;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
@@ -1510,8 +1509,14 @@ public class KafkaMessageChannelBinder extends
@Override
public void stop() {
if (this.producerFactory instanceof Lifecycle) {
((Lifecycle) producerFactory).stop();
if (this.producerFactory instanceof DisposableBean) {
try {
((DisposableBean) producerFactory).destroy();
}
catch (Exception ex) {
this.logger.error(ex, "Error destroying the producer factory bean: ");
throw new RuntimeException(ex);
}
}
this.running = false;
}