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; }