From cdc8ac0d6cdd7424c5ce4a8a641778328cd4013c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 1 Jul 2019 14:13:04 -0400 Subject: [PATCH] Fix race condition around Redis key https://build.spring.io/browse/INT-FATS5IC-922/ https://build.spring.io/browse/INT-MJATS41-1764/ The `RedisQueueMessageDrivenEndpointTests` may be called from different CI plans against the same Redis instance. So, clean up keys before and after every unit test **Cherry-pick until 4.3.x** # Conflicts: # spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java --- ...ueueMessageDrivenEndpointTests-context.xml | 7 +++- .../RedisQueueMessageDrivenEndpointTests.java | 39 +++++++++++++++++-- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml index c32aac1c79..361cdd497b 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml @@ -16,9 +16,10 @@ - @@ -28,8 +29,10 @@ - diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index c223d378d0..099aeafdcc 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -47,6 +47,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationEvent; +import org.springframework.context.Lifecycle; import org.springframework.data.redis.RedisConnectionFailureException; import org.springframework.data.redis.RedisSystemException; import org.springframework.data.redis.connection.RedisConnectionFactory; @@ -90,9 +91,15 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Autowired private PollableChannel fromChannel; + @Autowired + private Lifecycle fromChannelEndpoint; + @Autowired private MessageChannel symmetricalInputChannel; + @Autowired + private Lifecycle symmetricalRedisChannelEndpoint; + @Autowired private PollableChannel symmetricalOutputChannel; @@ -108,6 +115,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); String payload = "testing"; @@ -137,6 +145,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload2, receive.getPayload()); endpoint.stop(); + redisTemplate.delete(queueName); } @Test @@ -151,6 +160,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); Message message = MessageBuilder.withPayload("testing").build(); @@ -189,16 +199,21 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { Matchers.containsString("java.lang.String cannot be cast to org.springframework.messaging.Message")); endpoint.stop(); + redisTemplate.delete(queueName); } @Test @RedisAvailable public void testInt3017IntegrationInbound() { - String payload = new Date().toString(); - + String queueName = "si.test.redisQueueInboundChannelAdapterTests2"; RedisTemplate redisTemplate = new StringRedisTemplate(); redisTemplate.setConnectionFactory(this.connectionFactory); redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); + + this.fromChannelEndpoint.start(); + String payload = new Date().toString(); + redisTemplate.boundListOps("si.test.Int3017IntegrationInbound") .leftPush("{\"payload\":\"" + payload + "\",\"headers\":{}}"); @@ -206,14 +221,22 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { Message receive = this.fromChannel.receive(10000); assertNotNull(receive); assertEquals(payload, receive.getPayload()); + this.fromChannelEndpoint.stop(); + redisTemplate.delete(queueName); } @Test @RedisAvailable public void testInt3017IntegrationSymmetrical() { + String queueName = "si.test.Int3017IntegrationSymmetrical"; + RedisTemplate redisTemplate = new StringRedisTemplate(); + redisTemplate.setConnectionFactory(this.connectionFactory); + redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); + this.symmetricalRedisChannelEndpoint.start(); UUID payload = UUID.randomUUID(); Message message = MessageBuilder.withPayload(payload) - .setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical") + .setHeader("redis_queue", queueName) .build(); this.symmetricalInputChannel.send(message); @@ -221,6 +244,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { Message receive = this.symmetricalOutputChannel.receive(10000); assertNotNull(receive); assertEquals(payload, receive.getPayload()); + this.symmetricalRedisChannelEndpoint.stop(); + redisTemplate.delete(queueName); } @Test @@ -235,6 +260,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); while (redisTemplate.boundListOps(queueName).rightPop() != null) { // drain @@ -272,6 +298,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertTrue(stopLatch.await(21, TimeUnit.SECONDS)); verify(boundListOperations, atLeastOnce()).rightPush(any(byte[].class)); + redisTemplate.delete(queueName); } @@ -286,7 +313,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { final CountDownLatch exceptionsLatch = new CountDownLatch(2); - RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, + this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setApplicationEventPublisher(event -> { exceptionEvents.add((ApplicationEvent) event); @@ -329,6 +357,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload, receive.getPayload()); endpoint.stop(); + redisTemplate.delete(queueName); } @Test @@ -343,6 +372,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); redisTemplate.afterPropertiesSet(); + redisTemplate.delete(queueName); String payload = "testing"; @@ -373,6 +403,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { assertEquals(payload2, receive.getPayload()); endpoint.stop(); + redisTemplate.delete(queueName); } private void waitListening(RedisQueueMessageDrivenEndpoint endpoint) throws InterruptedException {