From f1e3a0bdd60ea90c9acd378f5351d355f30ecaff Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 30 Oct 2020 16:39:49 -0400 Subject: [PATCH] Allow retries in Kafka Streams binder (#980) * Allow retries in Kafka Streams binder Provide applications the capability to retry critical sections of the business logic. This is accomplished through a new API using which critical path can be wrapped inside a Callable. Adding tests and docs. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/945 * Reworking the PR Remove the binder API that was added before for retrying. Reuse binding provided RetryTemplate. Tests and docs * Cleanup * Addressing PR review comments * Fix typo --- docs/src/main/asciidoc/kafka-streams.adoc | 104 +++++++++ .../kafka/streams/GlobalKTableBinder.java | 14 +- .../GlobalKTableBinderConfiguration.java | 9 +- .../GlobalKTableBoundElementFactory.java | 11 +- .../binder/kafka/streams/KStreamBinder.java | 8 +- .../streams/KStreamBoundElementFactory.java | 3 +- .../binder/kafka/streams/KTableBinder.java | 14 +- .../streams/KTableBinderConfiguration.java | 8 +- .../streams/KTableBoundElementFactory.java | 11 +- ...StreamsBinderSupportAutoConfiguration.java | 12 +- .../streams/KafkaStreamsBinderUtils.java | 25 +- ...fkaStreamsBindingInformationCatalogue.java | 12 +- .../function/KafkaStreamsRetryTests.java | 216 ++++++++++++++++++ 13 files changed, 418 insertions(+), 29 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 40be05f94..4b0249444 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -926,6 +926,110 @@ This implies that if there are multiple functions or `StreamListener` methods in Unlike the support for deserialization exception handlers as described above, the binder does not provide such first class mechanisms for handling production exceptions. However, you still can configure production exception handlers using the `StreamsBuilderFactoryBean` customizer which you can find more details about, in a subsequent section below. +=== Retrying critical business logic + +There are scenarios in which you might want to retry parts of your business logic that are critical to the application. +There maybe an external call to a relational database or invoking a REST endpoint from the Kafka Streams processor. +These calls can fail for various reasons such as network issues or remote service unavailability. +More often, these failures may self resolve if you can try them again. +By default, Kafka Streams binder creates `RetryTemplate` beans for all the input bindings. + +If the function has the following signature, +``` +@Bean +public java.util.function.Consumer> process() +``` +and with default binding name, the `RetryTemplate` will be registered as `process-in-0-RetryTemplate`. +This is following the convention of binding name (`process-in-0`) followed by the literal `-RetryTemplate`. +In the case of multiple input bindings, there will be a separate `RetryTemplate` bean available per binding. +If there is a custom `RetryTemplate` bean available in the application and provided through `spring.cloud.stream.bindings..consumer.retryTemplateName`, then that takes precedence over any input binding level retry template configuration properties. + +Once the `RetryTemplate` from the binding is injected into the application, it can be used to retry any critical sections of the application. +Here is an example: + +``` +@Bean +public java.util.function.Consumer> process(@Lazy @Qualifier("process-in-0-RetryTemplate") RetryTemplate retryTemplate) { + + return input -> input + .process(() -> new Processor() { + @Override + public void init(ProcessorContext processorContext) { + } + + @Override + public void process(Object o, String s) { + retryTemplate.execute(context -> { + //Critical business logic goes here. + }); + } + + @Override + public void close() { + } + }); +} +``` + +Or you can use a custom `RetryTemplate` as below. + +``` +@EnableAutoConfiguration +public static class CustomRetryTemplateApp { + + @Bean + @StreamRetryTemplate + RetryTemplate fooRetryTemplate() { + RetryTemplate retryTemplate = new RetryTemplate(); + + RetryPolicy retryPolicy = new SimpleRetryPolicy(4); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1); + + retryTemplate.setBackOffPolicy(backOffPolicy); + retryTemplate.setRetryPolicy(retryPolicy); + + return retryTemplate; + } + + @Bean + public java.util.function.Consumer> process() { + + return input -> input + .process(() -> new Processor() { + @Override + public void init(ProcessorContext processorContext) { + } + + @Override + public void process(Object o, String s) { + fooRetryTemplate().execute(context -> { + //Critical business logic goes here. + }); + + } + + @Override + public void close() { + } + }); + } +} +``` + +Note that when retries are exhausted, by default, the last exception will be thrown, causing the processor to terminate. +If you wish to handle the exception and continue processing, you can add a RecoveryCallback to the `execute` method: +Here is an example. +``` +retryTemplate.execute(context -> { + //Critical business logic goes here. + }, context -> { + //Recovery logic goes here. + return null; + )); +``` +Refer to the https://github.com/spring-projects/spring-retry[Spring Retry] project for more information about the RetryTemplate, retry policies, backoff policies and more. + === State Store State stores are created automatically by Kafka Streams when the high level DSL is used and appropriate calls are made those trigger a state store. 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 ec99fe660..966456d83 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-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -30,6 +30,7 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.StringUtils; /** @@ -52,6 +53,8 @@ public class GlobalKTableBinder extends private final KafkaTopicProvisioner kafkaTopicProvisioner; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + // @checkstyle:off private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); @@ -59,9 +62,11 @@ public class GlobalKTableBinder extends public GlobalKTableBinder( KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - KafkaTopicProvisioner kafkaTopicProvisioner) { + KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; } @Override @@ -72,9 +77,12 @@ public class GlobalKTableBinder extends if (!StringUtils.hasText(group)) { group = properties.getExtension().getApplicationId(); } + final RetryTemplate retryTemplate = buildRetryTemplate(properties); KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties); + this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(), + this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget)); + 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/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java index e3e36349e..35e12240d 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -55,9 +55,11 @@ public class GlobalKTableBinderConfiguration { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, @Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties) { + GlobalKTableBinder globalKTableBinder = new GlobalKTableBinder(binderConfigurationProperties, - kafkaTopicProvisioner); + kafkaTopicProvisioner, kafkaStreamsBindingInformationCatalogue); globalKTableBinder.setKafkaStreamsExtendedBindingProperties( kafkaStreamsExtendedBindingProperties); return globalKTableBinder; @@ -76,6 +78,9 @@ public class GlobalKTableBinderConfiguration { beanFactory.registerSingleton( KafkaStreamsExtendedBindingProperties.class.getSimpleName(), outerContext.getBean(KafkaStreamsExtendedBindingProperties.class)); + beanFactory.registerSingleton( + KafkaStreamsBindingInformationCatalogue.class.getSimpleName(), + outerContext.getBean(KafkaStreamsBindingInformationCatalogue.class)); }; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java index 28037ed93..b7d125b57 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBoundElementFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -41,12 +41,15 @@ public class GlobalKTableBoundElementFactory private final BindingServiceProperties bindingServiceProperties; private final EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; GlobalKTableBoundElementFactory(BindingServiceProperties bindingServiceProperties, - EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { super(GlobalKTable.class); this.bindingServiceProperties = bindingServiceProperties; this.encodingDecodingBindAdviceHandler = encodingDecodingBindAdviceHandler; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; } @Override @@ -73,7 +76,9 @@ public class GlobalKTableBoundElementFactory GlobalKTable.class); proxyFactory.addAdvice(wrapper); - return (GlobalKTable) proxyFactory.getProxy(); + final GlobalKTable proxy = (GlobalKTable) proxyFactory.getProxy(); + this.kafkaStreamsBindingInformationCatalogue.addBindingNamePerTarget(proxy, name); + return proxy; } @Override 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 0ae19ef86..3835042de 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-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -38,6 +38,7 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.StringUtils; /** @@ -99,9 +100,12 @@ class KStreamBinder extends if (!StringUtils.hasText(group)) { group = properties.getExtension().getApplicationId(); } + + final RetryTemplate retryTemplate = buildRetryTemplate(properties); + KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties); + this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(), this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget)); 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/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java index dd6f78b6f..d484a5ba7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBoundElementFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2020 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. @@ -103,6 +103,7 @@ class KStreamBoundElementFactory extends AbstractBindingTargetFactory { .getBindingProperties(name); this.kafkaStreamsBindingInformationCatalogue.registerBindingProperties(proxy, bindingProperties); + this.kafkaStreamsBindingInformationCatalogue.addBindingNamePerTarget(proxy, name); return proxy; } 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 cb899e096..bf310f572 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-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -30,6 +30,7 @@ import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStr import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.StringUtils; /** @@ -53,15 +54,19 @@ class KTableBinder extends private final KafkaTopicProvisioner kafkaTopicProvisioner; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; + // @checkstyle:off private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); // @checkstyle:on KTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - KafkaTopicProvisioner kafkaTopicProvisioner) { + KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; + this.kafkaStreamsBindingInformationCatalogue = KafkaStreamsBindingInformationCatalogue; } @Override @@ -74,9 +79,12 @@ class KTableBinder extends if (!StringUtils.hasText(group)) { group = properties.getExtension().getApplicationId(); } + + final RetryTemplate retryTemplate = buildRetryTemplate(properties); KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties); + this.binderConfigurationProperties, properties, retryTemplate, getBeanFactory(), this.kafkaStreamsBindingInformationCatalogue.bindingNamePerTarget(inputTarget)); + 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/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index 75b5e0e48..4281bf653 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -55,9 +55,10 @@ public class KTableBinderConfiguration { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, @Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties) { KTableBinder kTableBinder = new KTableBinder(binderConfigurationProperties, - kafkaTopicProvisioner); + kafkaTopicProvisioner, kafkaStreamsBindingInformationCatalogue); kTableBinder.setKafkaStreamsExtendedBindingProperties(kafkaStreamsExtendedBindingProperties); return kTableBinder; } @@ -75,6 +76,9 @@ public class KTableBinderConfiguration { beanFactory.registerSingleton( KafkaStreamsExtendedBindingProperties.class.getSimpleName(), outerContext.getBean(KafkaStreamsExtendedBindingProperties.class)); + beanFactory.registerSingleton( + KafkaStreamsBindingInformationCatalogue.class.getSimpleName(), + outerContext.getBean(KafkaStreamsBindingInformationCatalogue.class)); }; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java index 7f45a794e..2dc7a90a7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBoundElementFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -39,12 +39,15 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { private final BindingServiceProperties bindingServiceProperties; private final EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler; + private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; KTableBoundElementFactory(BindingServiceProperties bindingServiceProperties, - EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { + EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler, + KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue) { super(KTable.class); this.bindingServiceProperties = bindingServiceProperties; this.encodingDecodingBindAdviceHandler = encodingDecodingBindAdviceHandler; + this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; } @Override @@ -68,7 +71,9 @@ class KTableBoundElementFactory extends AbstractBindingTargetFactory { KTableBoundElementFactory.KTableWrapper.class, KTable.class); proxyFactory.addAdvice(wrapper); - return (KTable) proxyFactory.getProxy(); + final KTable proxy = (KTable) proxyFactory.getProxy(); + this.kafkaStreamsBindingInformationCatalogue.addBindingNamePerTarget(proxy, name); + return proxy; } @Override diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index c29ecd851..7c2787bc6 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2020 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. @@ -345,14 +345,16 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean public KTableBoundElementFactory kTableBoundElementFactory( - BindingServiceProperties bindingServiceProperties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { - return new KTableBoundElementFactory(bindingServiceProperties, encodingDecodingBindAdviceHandler); + BindingServiceProperties bindingServiceProperties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler, + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { + return new KTableBoundElementFactory(bindingServiceProperties, encodingDecodingBindAdviceHandler, KafkaStreamsBindingInformationCatalogue); } @Bean public GlobalKTableBoundElementFactory globalKTableBoundElementFactory( - BindingServiceProperties properties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler) { - return new GlobalKTableBoundElementFactory(properties, encodingDecodingBindAdviceHandler); + BindingServiceProperties properties, EncodingDecodingBindAdviceHandler encodingDecodingBindAdviceHandler, + KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue) { + return new GlobalKTableBoundElementFactory(properties, encodingDecodingBindAdviceHandler, KafkaStreamsBindingInformationCatalogue); } @Bean 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 92f10a1a5..e57fa0e4d 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-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -28,6 +28,10 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.streams.kstream.KStream; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; @@ -45,6 +49,7 @@ import org.springframework.kafka.core.KafkaOperations; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -64,9 +69,11 @@ final class KafkaStreamsBinderUtils { } static void prepareConsumerBinding(String name, String group, - ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner, - KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - ExtendedConsumerProperties properties) { + ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner, + KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, + ExtendedConsumerProperties properties, + RetryTemplate retryTemplate, + ConfigurableListableBeanFactory beanFactory, String bindingName) { ExtendedConsumerProperties extendedConsumerProperties = (ExtendedConsumerProperties) properties; @@ -131,6 +138,16 @@ final class KafkaStreamsBinderUtils { kafkaStreamsBinderDlqRecoverer); } } + + if (StringUtils.hasText(properties.getRetryTemplateName())) { + @SuppressWarnings("unchecked") + BeanDefinition retryTemplateBeanDefinition = BeanDefinitionBuilder + .genericBeanDefinition( + (Class) retryTemplate.getClass(), + () -> retryTemplate) + .getRawBeanDefinition(); + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(bindingName + "-RetryTemplate", retryTemplateBeanDefinition); + } } private static DefaultKafkaProducerFactory getProducerFactory( 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 56d91c64a..540d63b19 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-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -53,6 +53,8 @@ class KafkaStreamsBindingInformationCatalogue { private final Map, Serde> keySerdeInfo = new HashMap<>(); + private final Map bindingNamesPerTarget = new HashMap<>(); + /** * For a given bounded {@link KStream}, retrieve it's corresponding destination on the * broker. @@ -161,4 +163,12 @@ class KafkaStreamsBindingInformationCatalogue { Map, KafkaStreamsConsumerProperties> getConsumerProperties() { return consumerProperties; } + + void addBindingNamePerTarget(Object target, String bindingName) { + this.bindingNamesPerTarget.put(target, bindingName); + } + + String bindingNamePerTarget(Object target) { + return this.bindingNamesPerTarget.get(target); + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java new file mode 100644 index 000000000..285822314 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java @@ -0,0 +1,216 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams.function; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; + +import org.apache.kafka.streams.kstream.GlobalKTable; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; +import org.apache.kafka.streams.processor.Processor; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.beans.factory.NoSuchBeanDefinitionException; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.StreamRetryTemplate; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Lazy; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.retry.RetryPolicy; +import org.springframework.retry.backoff.FixedBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.util.Assert; + +import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; +import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat; + +public class KafkaStreamsRetryTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true); + + private static final EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + private final static CountDownLatch LATCH1 = new CountDownLatch(2); + private final static CountDownLatch LATCH2 = new CountDownLatch(4); + + @Test + public void testRetryTemplatePerBindingOnKStream() throws Exception { + SpringApplication app = new SpringApplication(RetryTemplatePerConsumerBindingApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.function.definition=process", + "--spring.cloud.stream.bindings.process-in-0.destination=words", + "--spring.cloud.stream.bindings.process-in-0.consumer.max-attempts=2", + "--spring.cloud.stream.kafka.streams.default.consumer.application-id=testRetryTemplatePerBindingOnKStream", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + sendAndValidate(LATCH1); + } + } + + @Test + public void testRetryTemplateOnTableTypes() throws Exception { + SpringApplication app = new SpringApplication(RetryTemplatePerConsumerBindingApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.function.definition=tableTypes", + "--spring.cloud.stream.kafka.streams.default.consumer.application-id=testRetryTemplateOnTableTypes", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + assertThat(context.getBean("tableTypes-in-0-RetryTemplate", RetryTemplate.class)).isNotNull(); + assertThat(context.getBean("tableTypes-in-1-RetryTemplate", RetryTemplate.class)).isNotNull(); + } + } + + @Test + public void testRetryTemplateBeanProvidedByTheApp() throws Exception { + SpringApplication app = new SpringApplication(CustomRetryTemplateApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.function.definition=process", + "--spring.cloud.stream.bindings.process-in-0.destination=words", + "--spring.cloud.stream.bindings.process-in-0.consumer.retry-template-name=fooRetryTemplate", + "--spring.cloud.stream.kafka.streams.default.consumer.application-id=testRetryTemplateBeanProvidedByTheApp", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + sendAndValidate(LATCH2); + assertThatThrownBy(() -> context.getBean("process-in-0-RetryTemplate", RetryTemplate.class)).isInstanceOf(NoSuchBeanDefinitionException.class); + } + } + + private void sendAndValidate(CountDownLatch latch) throws InterruptedException { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("foobar"); + Assert.isTrue(latch.await(10, TimeUnit.SECONDS), "Foo"); + } + finally { + pf.destroy(); + } + } + + @EnableAutoConfiguration + public static class RetryTemplatePerConsumerBindingApp { + + @Bean + public java.util.function.Consumer> process(@Lazy @Qualifier("process-in-0-RetryTemplate") RetryTemplate retryTemplate) { + + return input -> input + .process(() -> new Processor() { + @Override + public void init(ProcessorContext processorContext) { + } + + @Override + public void process(Object o, String s) { + retryTemplate.execute(context -> { + LATCH1.countDown(); + throw new RuntimeException(); + }); + } + + @Override + public void close() { + } + }); + } + + @Bean + public BiConsumer, GlobalKTable> tableTypes() { + return (t, g) -> { + }; + } + } + + @EnableAutoConfiguration + public static class CustomRetryTemplateApp { + + @Bean + @StreamRetryTemplate + RetryTemplate fooRetryTemplate() { + RetryTemplate retryTemplate = new RetryTemplate(); + + RetryPolicy retryPolicy = new SimpleRetryPolicy(4); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1); + + retryTemplate.setBackOffPolicy(backOffPolicy); + retryTemplate.setRetryPolicy(retryPolicy); + + return retryTemplate; + } + + @Bean + public java.util.function.Consumer> process() { + + return input -> input + .process(() -> new Processor() { + @Override + public void init(ProcessorContext processorContext) { + } + + @Override + public void process(Object o, String s) { + fooRetryTemplate().execute(context -> { + LATCH2.countDown(); + throw new RuntimeException(); + }); + + } + + @Override + public void close() { + } + }); + } + } +}