Skip to content

Commit 3203058

Browse files
tomazfernandestomazfernandes
authored
GH-1745 - Improve RetryTopicConfigurer flexibility (#1821)
* GH-1745 - Improve RetryTopicConfigurer flexibility * GH-1745 - Implement code review suggestions * GH-1745 - Make EndpointCustomizer public
1 parent d1e0450 commit 3203058

File tree

4 files changed

+200
-104
lines changed

4 files changed

+200
-104
lines changed

‎spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopic.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,14 @@ public boolean shouldRetryOn(Integer attempt, Throwable e) {
8686
return this.properties.shouldRetryOn.test(attempt, e);
8787
}
8888

89+
@Override
90+
public String toString() {
91+
return "DestinationTopic{" +
92+
"destinationName='" + this.destinationName + '\'' +
93+
", properties=" + this.properties +
94+
'}';
95+
}
96+
8997
@Override
9098
public boolean equals(Object o) {
9199
if (this == o) {
@@ -191,6 +199,21 @@ public int hashCode() {
191199
this.dltStrategy, this.kafkaOperations);
192200
}
193201

202+
@Override
203+
public String toString() {
204+
return "Properties{" +
205+
"delayMs=" + this.delayMs +
206+
", suffix='" + this.suffix + '\'' +
207+
", type=" + this.type +
208+
", maxAttempts=" + this.maxAttempts +
209+
", numPartitions=" + this.numPartitions +
210+
", dltStrategy=" + this.dltStrategy +
211+
", kafkaOperations=" + this.kafkaOperations +
212+
", shouldRetryOn=" + this.shouldRetryOn +
213+
", timeout=" + this.timeout +
214+
'}';
215+
}
216+
194217
public boolean isMainEndpoint() {
195218
return Type.MAIN.equals(this.type);
196219
}

‎spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizer.java‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
/*
2+
* Copyright 2021 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.kafka.retrytopic;
18+
19+
import java.util.Collection;
20+
21+
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
22+
23+
/**
24+
* Customizes main, retry and DLT endpoints in the Retry Topic functionality
25+
* and returns the resulting topic names.
26+
*
27+
* @author Tomaz Fernandes
28+
* @since 2.7.2
29+
*
30+
* @see EndpointCustomizerFactory
31+
*
32+
*/
33+
@FunctionalInterface
34+
public interface EndpointCustomizer {
35+
36+
/**
37+
* Customize the endpoint and return the topic names generated for this endpoint.
38+
* @param listenerEndpoint The main, retry or DLT endpoint to be customized.
39+
* @return A collection containing the topic names generated for this endpoint.
40+
*/
41+
Collection<TopicNamesHolder> customizeEndpointAndCollectTopics(MethodKafkaListenerEndpoint<?, ?> listenerEndpoint);
42+
43+
class TopicNamesHolder {
44+
45+
private final String mainTopic;
46+
47+
private final String customizedTopic;
48+
49+
TopicNamesHolder(String mainTopic, String customizedTopic) {
50+
this.mainTopic = mainTopic;
51+
this.customizedTopic = customizedTopic;
52+
}
53+
54+
String getMainTopic() {
55+
return this.mainTopic;
56+
}
57+
58+
String getCustomizedTopic() {
59+
return this.customizedTopic;
60+
}
61+
}
62+
}

‎spring-kafka/src/main/java/org/springframework/kafka/retrytopic/EndpointCustomizerFactory.java‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
/*
2+
* Copyright 2018-2021 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.kafka.retrytopic;
18+
19+
import java.lang.reflect.Method;
20+
import java.util.Arrays;
21+
import java.util.Collection;
22+
import java.util.stream.Collectors;
23+
24+
import org.springframework.beans.factory.BeanFactory;
25+
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
26+
import org.springframework.kafka.support.TopicPartitionOffset;
27+
28+
/**
29+
*
30+
* Creates the {@link EndpointCustomizer} that will be used by the {@link RetryTopicConfigurer}
31+
* to customize the main, retry and DLT endpoints.
32+
*
33+
* @author Tomaz Fernandes
34+
* @since 2.7.2
35+
*
36+
* @see RetryTopicConfigurer
37+
* @see DestinationTopic.Properties
38+
* @see RetryTopicNamesProviderFactory.RetryTopicNamesProvider
39+
*
40+
*/
41+
public class EndpointCustomizerFactory {
42+
43+
private final DestinationTopic.Properties destinationProperties;
44+
45+
private final EndpointHandlerMethod beanMethod;
46+
47+
private final BeanFactory beanFactory;
48+
49+
private final RetryTopicNamesProviderFactory retryTopicNamesProviderFactory;
50+
51+
EndpointCustomizerFactory(DestinationTopic.Properties destinationProperties, EndpointHandlerMethod beanMethod,
52+
BeanFactory beanFactory, RetryTopicNamesProviderFactory retryTopicNamesProviderFactory) {
53+
54+
this.destinationProperties = destinationProperties;
55+
this.beanMethod = beanMethod;
56+
this.beanFactory = beanFactory;
57+
this.retryTopicNamesProviderFactory = retryTopicNamesProviderFactory;
58+
}
59+
60+
final public EndpointCustomizer createEndpointCustomizer() {
61+
return addSuffixesAndMethod(this.destinationProperties, this.beanMethod.resolveBean(this.beanFactory),
62+
this.beanMethod.getMethod());
63+
}
64+
65+
protected EndpointCustomizer addSuffixesAndMethod(DestinationTopic.Properties properties, Object bean, Method method) {
66+
RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider =
67+
this.retryTopicNamesProviderFactory.createRetryTopicNamesProvider(properties);
68+
return endpoint -> {
69+
Collection<EndpointCustomizer.TopicNamesHolder> topics = customizeAndRegisterTopics(namesProvider, endpoint);
70+
endpoint.setId(namesProvider.getEndpointId(endpoint));
71+
endpoint.setGroupId(namesProvider.getGroupId(endpoint));
72+
endpoint.setTopics(topics.stream().map(EndpointCustomizer.TopicNamesHolder::getCustomizedTopic).toArray(String[]::new));
73+
endpoint.setClientIdPrefix(namesProvider.getClientIdPrefix(endpoint));
74+
endpoint.setGroup(namesProvider.getGroup(endpoint));
75+
endpoint.setBean(bean);
76+
endpoint.setMethod(method);
77+
return topics;
78+
};
79+
}
80+
81+
protected Collection<EndpointCustomizer.TopicNamesHolder> customizeAndRegisterTopics(
82+
RetryTopicNamesProviderFactory.RetryTopicNamesProvider namesProvider,
83+
MethodKafkaListenerEndpoint<?, ?> endpoint) {
84+
85+
return getTopics(endpoint)
86+
.stream()
87+
.map(topic -> new EndpointCustomizer.TopicNamesHolder(topic, namesProvider.getTopicName(topic)))
88+
.collect(Collectors.toList());
89+
}
90+
91+
private Collection<String> getTopics(MethodKafkaListenerEndpoint<?, ?> endpoint) {
92+
Collection<String> topics = endpoint.getTopics();
93+
if (topics.isEmpty()) {
94+
TopicPartitionOffset[] topicPartitionsToAssign = endpoint.getTopicPartitionsToAssign();
95+
if (topicPartitionsToAssign != null && topicPartitionsToAssign.length > 0) {
96+
topics = Arrays.stream(topicPartitionsToAssign)
97+
.map(TopicPartitionOffset::getTopic)
98+
.collect(Collectors.toList());
99+
}
100+
}
101+
102+
if (topics.isEmpty()) {
103+
throw new IllegalStateException(
104+
String.format("No topics were provided for RetryTopicConfiguration for method %s in class %s.",
105+
endpoint.getMethod().getName(), endpoint.getBean().getClass().getSimpleName()));
106+
}
107+
return topics;
108+
}
109+
}

‎spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java‎

Lines changed: 6 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,8 @@
1717
package org.springframework.kafka.retrytopic;
1818

1919
import java.lang.reflect.Method;
20-
import java.util.Arrays;
2120
import java.util.Collection;
2221
import java.util.function.Consumer;
23-
import java.util.function.Function;
24-
import java.util.stream.Collectors;
2522

2623
import org.apache.commons.logging.LogFactory;
2724
import org.apache.kafka.clients.admin.NewTopic;
@@ -37,8 +34,6 @@
3734
import org.springframework.kafka.config.MethodKafkaListenerEndpoint;
3835
import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
3936
import org.springframework.kafka.listener.ListenerUtils;
40-
import org.springframework.kafka.retrytopic.RetryTopicNamesProviderFactory.RetryTopicNamesProvider;
41-
import org.springframework.kafka.support.TopicPartitionOffset;
4237
import org.springframework.lang.Nullable;
4338

4439

@@ -284,7 +279,7 @@ private void configureEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
284279
String defaultContainerFactoryBeanName) {
285280
this.destinationTopicProcessor
286281
.processDestinationTopicProperties(destinationTopicProperties ->
287-
processAndRegisterEndpoints(mainEndpoint,
282+
processAndRegisterEndpoint(mainEndpoint,
288283
endpointProcessor,
289284
factory,
290285
defaultContainerFactoryBeanName,
@@ -295,7 +290,7 @@ private void configureEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
295290
context);
296291
}
297292

298-
private void processAndRegisterEndpoints(MethodKafkaListenerEndpoint<?, ?> mainEndpoint, EndpointProcessor endpointProcessor,
293+
private void processAndRegisterEndpoint(MethodKafkaListenerEndpoint<?, ?> mainEndpoint, EndpointProcessor endpointProcessor,
299294
KafkaListenerContainerFactory<?> factory,
300295
String defaultFactoryBeanName,
301296
KafkaListenerEndpointRegistrar registrar,
@@ -321,14 +316,14 @@ private void processAndRegisterEndpoints(MethodKafkaListenerEndpoint<?, ?> mainE
321316
.forEach(topicNamesHolder ->
322317
this.destinationTopicProcessor
323318
.registerDestinationTopic(topicNamesHolder.getMainTopic(),
324-
topicNamesHolder.getProcessedTopic(),
319+
topicNamesHolder.getCustomizedTopic(),
325320
destinationTopicProperties, context));
326321

327322
registrar.registerEndpoint(endpoint, resolvedFactory);
328323
endpoint.setBeanFactory(this.beanFactory);
329324
}
330325

331-
private EndpointHandlerMethod getEndpointHandlerMethod(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
326+
protected EndpointHandlerMethod getEndpointHandlerMethod(MethodKafkaListenerEndpoint<?, ?> mainEndpoint,
332327
RetryTopicConfiguration configuration,
333328
DestinationTopic.Properties props) {
334329
EndpointHandlerMethod dltHandlerMethod = configuration.getDltHandlerMethod();
@@ -343,15 +338,15 @@ private Consumer<Collection<String>> getTopicCreationFunction(RetryTopicConfigur
343338
: topics -> { };
344339
}
345340

346-
private void createNewTopicBeans(Collection<String> topics, RetryTopicConfiguration.TopicCreation config) {
341+
protected void createNewTopicBeans(Collection<String> topics, RetryTopicConfiguration.TopicCreation config) {
347342
topics.forEach(topic ->
348343
((DefaultListableBeanFactory) this.beanFactory)
349344
.registerSingleton(topic + "-topicRegistrationBean",
350345
new NewTopic(topic, config.getNumPartitions(), config.getReplicationFactor()))
351346
);
352347
}
353348

354-
private EndpointCustomizer createEndpointCustomizer(
349+
protected EndpointCustomizer createEndpointCustomizer(
355350
EndpointHandlerMethod endpointBeanMethod, DestinationTopic.Properties destinationTopicProperties) {
356351

357352
return new EndpointCustomizerFactory(destinationTopicProperties,
@@ -407,99 +402,6 @@ default void process(MethodKafkaListenerEndpoint<?, ?> listenerEndpoint) {
407402
}
408403
}
409404

410-
private interface EndpointCustomizer extends Function<MethodKafkaListenerEndpoint<?, ?>, Collection<TopicNamesHolder>> {
411-
default Collection<TopicNamesHolder> customizeEndpointAndCollectTopics(MethodKafkaListenerEndpoint<?, ?> listenerEndpoint) {
412-
return apply(listenerEndpoint);
413-
}
414-
}
415-
416-
417-
static final class EndpointCustomizerFactory {
418-
419-
private final DestinationTopic.Properties destinationProperties;
420-
421-
private final EndpointHandlerMethod beanMethod;
422-
423-
private final BeanFactory beanFactory;
424-
425-
private final RetryTopicNamesProviderFactory retryTopicNamesProviderFactory;
426-
427-
EndpointCustomizerFactory(DestinationTopic.Properties destinationProperties, EndpointHandlerMethod beanMethod,
428-
BeanFactory beanFactory, RetryTopicNamesProviderFactory retryTopicNamesProviderFactory) {
429-
430-
this.destinationProperties = destinationProperties;
431-
this.beanMethod = beanMethod;
432-
this.beanFactory = beanFactory;
433-
this.retryTopicNamesProviderFactory = retryTopicNamesProviderFactory;
434-
}
435-
436-
public EndpointCustomizer createEndpointCustomizer() {
437-
return addSuffixesAndMethod(this.destinationProperties, this.beanMethod.resolveBean(this.beanFactory),
438-
this.beanMethod.getMethod());
439-
}
440-
441-
private EndpointCustomizer addSuffixesAndMethod(DestinationTopic.Properties properties, Object bean, Method method) {
442-
RetryTopicNamesProvider namesProvider = this.retryTopicNamesProviderFactory.createRetryTopicNamesProvider(properties);
443-
return endpoint -> {
444-
Collection<TopicNamesHolder> topics = customizeAndRegisterTopics(namesProvider, endpoint);
445-
endpoint.setId(namesProvider.getEndpointId(endpoint));
446-
endpoint.setGroupId(namesProvider.getGroupId(endpoint));
447-
endpoint.setTopics(topics.stream().map(TopicNamesHolder::getProcessedTopic).toArray(String[]::new));
448-
endpoint.setClientIdPrefix(namesProvider.getClientIdPrefix(endpoint));
449-
endpoint.setGroup(namesProvider.getGroup(endpoint));
450-
endpoint.setBean(bean);
451-
endpoint.setMethod(method);
452-
return topics;
453-
};
454-
}
455-
456-
private Collection<TopicNamesHolder> customizeAndRegisterTopics(RetryTopicNamesProvider namesProvider,
457-
MethodKafkaListenerEndpoint<?, ?> endpoint) {
458-
459-
return getTopics(endpoint)
460-
.stream()
461-
.map(topic -> new TopicNamesHolder(topic, namesProvider.getTopicName(topic)))
462-
.collect(Collectors.toList());
463-
}
464-
465-
private Collection<String> getTopics(MethodKafkaListenerEndpoint<?, ?> endpoint) {
466-
Collection<String> topics = endpoint.getTopics();
467-
if (topics.isEmpty()) {
468-
TopicPartitionOffset[] topicPartitionsToAssign = endpoint.getTopicPartitionsToAssign();
469-
if (topicPartitionsToAssign != null && topicPartitionsToAssign.length > 0) {
470-
topics = Arrays.stream(topicPartitionsToAssign)
471-
.map(TopicPartitionOffset::getTopic)
472-
.collect(Collectors.toList());
473-
}
474-
}
475-
476-
if (topics.isEmpty()) {
477-
throw new IllegalStateException("No topics where provided for RetryTopicConfiguration.");
478-
}
479-
return topics;
480-
}
481-
}
482-
483-
private static final class TopicNamesHolder {
484-
485-
private final String mainTopic;
486-
487-
private final String processedTopic;
488-
489-
TopicNamesHolder(String mainTopic, String processedTopic) {
490-
this.mainTopic = mainTopic;
491-
this.processedTopic = processedTopic;
492-
}
493-
494-
String getMainTopic() {
495-
return this.mainTopic;
496-
}
497-
498-
String getProcessedTopic() {
499-
return this.processedTopic;
500-
}
501-
}
502-
503405
static class LoggingDltListenerHandlerMethod {
504406

505407
public static final String DEFAULT_DLT_METHOD_NAME = "logMessage";

0 commit comments

Comments
 (0)