From e2eca34e4b8f7fc0dab2f19a19094cce0dfda2e4 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 2 Mar 2021 16:24:34 -0500 Subject: [PATCH] Cleaning up Kafka producer factories 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 --- .../kafka/streams/GlobalKTableBinder.java | 5 +++-- .../binder/kafka/streams/KStreamBinder.java | 6 ++++-- .../binder/kafka/streams/KTableBinder.java | 6 ++++-- .../kafka/streams/KafkaStreamsBinderUtils.java | 7 +++++-- ...KafkaStreamsBindingInformationCatalogue.java | 17 +++++++++++++++-- .../streams/StreamsBuilderFactoryManager.java | 7 ++++++- .../DlqDestinationResolverTests.java | 8 ++++++-- .../binder/kafka/KafkaMessageChannelBinder.java | 13 +++++++++---- 8 files changed, 52 insertions(+), 17 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java index 966456d83..808b1cd0b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java @@ -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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index f6580a971..6b86ad290 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index bf310f572..31fbba90b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -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); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index f1c1bb831..b07b97a3f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -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 properties, RetryTemplate retryTemplate, - ConfigurableListableBeanFactory beanFactory, String bindingName) { + ConfigurableListableBeanFactory beanFactory, String bindingName, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { ExtendedConsumerProperties extendedConsumerProperties = (ExtendedConsumerProperties) properties; @@ -109,6 +110,8 @@ final class KafkaStreamsBinderUtils { new ExtendedProducerProperties<>( extendedConsumerProperties.getExtension().getDlqProducerProperties()), binderConfigurationProperties); + kafkaStreamsBindingInformationCatalogue.addDlqProducerFactory(producerFactory); + KafkaOperations kafkaTemplate = new KafkaTemplate<>(producerFactory); Map dlqDestinationResolvers = diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java index ccf441f47..9b240b01b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBindingInformationCatalogue.java @@ -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, BindingProperties> bindingProperties = new ConcurrentHashMap<>(); @@ -55,6 +58,8 @@ class KafkaStreamsBindingInformationCatalogue { private final Map bindingNamesPerTarget = new HashMap<>(); + private final List> 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> getDlqProducerFactories() { + return this.dlqProducerFactories; + } + + public void addDlqProducerFactory(ProducerFactory producerFactory) { + this.dlqProducerFactories.add(producerFactory); + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java index 2c9d15bc3..bbea2c38e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/StreamsBuilderFactoryManager.java @@ -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 dlqProducerFactory : this.kafkaStreamsBindingInformationCatalogue.getDlqProducerFactories()) { + ((DisposableBean) dlqProducerFactory).destroy(); + } } catch (Exception ex) { throw new IllegalStateException(ex); diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java index 958c2d823..e45bc639a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java @@ -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 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(); diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index beae1040a..7882ab9c4 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -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; }