Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2477: Fix Static State RetryTopicConfigSupport

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/2477

Previously, a static field was used to detect/prevent multiple instances
of `RetryTopicConfigurationSupport`. This causes problems in test suites
or when an application context might be destroyed and re-created in the
same class loader.

Instead, detect and warn when multiple instances are found in an application context.

**cherry-pick to 2.9.x**

* Fix test.
This commit is contained in:
Gary Russell
2022-11-14 12:32:36 -05:00
committed by GitHub
parent f0cba91a8f
commit 2f9a7f758e
13 changed files with 62 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java
View File

@@ -19,17 +19,22 @@ package org.springframework.kafka.retrytopic;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.Map;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.annotation.EnableKafkaRetryTopic;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.kafka.config.KafkaListenerConfigUtils;
@@ -67,14 +72,29 @@ import org.springframework.util.backoff.FixedBackOff;
* @author Gary Russell
* @since 2.9
*/
public class RetryTopicConfigurationSupport {
private static final AtomicBoolean ONLY_ONE_ALLOWED = new AtomicBoolean(true);
public class RetryTopicConfigurationSupport implements ApplicationContextAware, SmartInitializingSingleton {
private final RetryTopicComponentFactory componentFactory = createComponentFactory();
public RetryTopicConfigurationSupport() {
Assert.state(ONLY_ONE_ALLOWED.getAndSet(false), "Only one 'RetryTopicConfigurationSupport' is allowed");
private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
private ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public void afterSingletonsInstantiated() {
if (this.applicationContext != null) {
Map<String, RetryTopicConfigurationSupport> beans = this.applicationContext
.getBeansOfType(RetryTopicConfigurationSupport.class, false, false);
if (beans.size() > 1) {
this.logger.warn(() -> "Only one RetryTopicConfigurationSupport object expected, found "
+ beans.keySet() + "; this may result in unexpected behavior");
}
}
}
/**

41
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AbstractRetryTopicIntegrationTests.java
View File

@@ -1,41 +0,0 @@
/*
* Copyright 2022 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.kafka.retrytopic;
import java.lang.reflect.Field;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.BeforeAll;
/**
* @author Gary Russell
* @since 2.9
*
*/
public class AbstractRetryTopicIntegrationTests {
protected AbstractRetryTopicIntegrationTests() {
}
@BeforeAll
static void reset() throws Exception { // NOSONAR
Field field = RetryTopicConfigurationSupport.class.getDeclaredField("ONLY_ONE_ALLOWED");
field.setAccessible(true);
((AtomicBoolean) field.get(null)).set(true);
}
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/CircularDltHandlerTests.java
View File

@@ -36,7 +36,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
* @since 2.8
*
*/
public class CircularDltHandlerTests extends AbstractRetryTopicIntegrationTests {
public class CircularDltHandlerTests {
@Test
void contextLoads() {

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DeliveryHeaderTests.java
View File

@@ -58,7 +58,7 @@ import org.springframework.util.backoff.FixedBackOff;
@SpringJUnitConfig
@DirtiesContext
@EmbeddedKafka(topics = "dh1")
public class DeliveryHeaderTests extends AbstractRetryTopicIntegrationTests {
public class DeliveryHeaderTests {
@Test
void deliveryAttempts(@Autowired Config config, @Autowired KafkaTemplate<Integer, String> template)

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DltStartupTests.java
View File

@@ -53,7 +53,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
*/
@SpringJUnitConfig
@EmbeddedKafka
public class DltStartupTests extends AbstractRetryTopicIntegrationTests {
public class DltStartupTests {
@Test
void dltStartOverridesCorrect(@Autowired KafkaListenerEndpointRegistry registry) {

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ExistingRetryTopicIntegrationTests.java
View File

@@ -76,7 +76,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
ExistingRetryTopicIntegrationTests.MAIN_TOPIC_WITH_PARTITION_INFO,
ExistingRetryTopicIntegrationTests.RETRY_TOPIC_WITH_PARTITION_INFO}, partitions = 4)
@TestPropertySource(properties = "two.attempts=2")
public class ExistingRetryTopicIntegrationTests extends AbstractRetryTopicIntegrationTests {
public class ExistingRetryTopicIntegrationTests {
private static final Logger logger = LoggerFactory.getLogger(ExistingRetryTopicIntegrationTests.class);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/PartitionResolverTests.java
View File

@@ -60,7 +60,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@DirtiesContext
@SpringJUnitConfig
@EmbeddedKafka(topics = "partition.resolver.tests")
public class PartitionResolverTests extends AbstractRetryTopicIntegrationTests {
public class PartitionResolverTests {
@Test
void testNullPartition(@Autowired KafkaOperations<Integer, String> template,

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java
View File

@@ -57,7 +57,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@SpringJUnitConfig
@DirtiesContext
@EmbeddedKafka(topics = RetryTopicConfigurationIntegrationTests.TOPIC1, partitions = 1)
class RetryTopicConfigurationIntegrationTests extends AbstractRetryTopicIntegrationTests {
class RetryTopicConfigurationIntegrationTests {
public static final String TOPIC1 = "RetryTopicConfigurationIntegrationTests.1";

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java
View File

@@ -60,7 +60,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@DirtiesContext
@EmbeddedKafka(topics = { RetryTopicConfigurationManualAssignmentIntegrationTests.TOPIC1,
RetryTopicConfigurationManualAssignmentIntegrationTests.TOPIC2 }, partitions = 1)
class RetryTopicConfigurationManualAssignmentIntegrationTests extends AbstractRetryTopicIntegrationTests {
class RetryTopicConfigurationManualAssignmentIntegrationTests {
public static final String TOPIC1 = "RetryTopicConfigurationManualAssignmentIntegrationTests.1";

36
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupportTests.java
View File

@@ -22,20 +22,23 @@ import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.then;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.lang.reflect.Field;
import java.time.Clock;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.log.LogAccessor;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerPartitionPausingBackOffManagerFactory;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
@@ -43,6 +46,7 @@ import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
import org.springframework.kafka.listener.ListenerContainerRegistry;
import org.springframework.kafka.support.converter.ConversionException;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.backoff.BackOff;
@@ -53,13 +57,6 @@ import org.springframework.util.backoff.BackOff;
*/
class RetryTopicConfigurationSupportTests {
@BeforeEach
void reset() throws Exception { // NOSONAR
Field field = RetryTopicConfigurationSupport.class.getDeclaredField("ONLY_ONE_ALLOWED");
field.setAccessible(true);
((AtomicBoolean) field.get(null)).set(true);
}
@SuppressWarnings("unchecked")
@Test
void testCreateConfigurer() {
@@ -242,4 +239,23 @@ class RetryTopicConfigurationSupportTests {
assertThat(configurationSupport).hasFieldOrProperty("componentFactory").isNotNull();
}
@Test
void twoSupports() {
RetryTopicConfigurationSupport configurationSupport = new RetryTopicConfigurationSupport();
LogAccessor logger = spy(KafkaTestUtils.getPropertyValue(configurationSupport, "logger", LogAccessor.class));
new DirectFieldAccessor(configurationSupport).setPropertyValue("logger", logger);
ApplicationContext ctx = mock(ApplicationContext.class);
configurationSupport.setApplicationContext(ctx);
Map<String, RetryTopicConfigurationSupport> beans = new LinkedHashMap<>();
beans.put("foo", configurationSupport);
beans.put("bar", configurationSupport);
given(ctx.getBeansOfType(RetryTopicConfigurationSupport.class, false, false)).willReturn(beans);
configurationSupport.afterSingletonsInstantiated();
@SuppressWarnings("unchecked")
ArgumentCaptor<Supplier<String>> captor = ArgumentCaptor.forClass(Supplier.class);
verify(logger).warn(captor.capture());
assertThat(captor.getValue().get()).isEqualTo("Only one RetryTopicConfigurationSupport object expected, found "
+ "[foo, bar]; this may result in unexpected behavior");
}
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicExceptionRoutingIntegrationTests.java
View File

@@ -70,7 +70,7 @@ import org.springframework.util.backoff.FixedBackOff;
@SpringJUnitConfig
@DirtiesContext
@EmbeddedKafka
public class RetryTopicExceptionRoutingIntegrationTests extends AbstractRetryTopicIntegrationTests {
public class RetryTopicExceptionRoutingIntegrationTests {
private static final Logger logger = LoggerFactory.getLogger(RetryTopicExceptionRoutingIntegrationTests.class);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicIntegrationTests.java
View File

@@ -96,7 +96,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
RetryTopicIntegrationTests.FOURTH_TOPIC,
RetryTopicIntegrationTests.TWO_LISTENERS_TOPIC })
@TestPropertySource(properties = { "five.attempts=5", "kafka.template=customKafkaTemplate"})
public class RetryTopicIntegrationTests extends AbstractRetryTopicIntegrationTests {
public class RetryTopicIntegrationTests {
private static final Logger logger = LoggerFactory.getLogger(RetryTopicIntegrationTests.class);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicSameContainerFactoryIntegrationTests.java
View File

@@ -67,7 +67,7 @@ import org.springframework.util.backoff.FixedBackOff;
@DirtiesContext
@EmbeddedKafka(topics = { RetryTopicSameContainerFactoryIntegrationTests.FIRST_TOPIC,
RetryTopicSameContainerFactoryIntegrationTests.SECOND_TOPIC}, partitions = 1)
public class RetryTopicSameContainerFactoryIntegrationTests extends AbstractRetryTopicIntegrationTests {
public class RetryTopicSameContainerFactoryIntegrationTests {
private static final Logger logger = LoggerFactory.getLogger(RetryTopicSameContainerFactoryIntegrationTests.class);