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

# Conflicts:
#	spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

# Conflicts:
#	spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml
#	spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

Author: artembilan

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml

@@ -18,10 +18,11 @@
 		<int:queue/>
 	</int:channel>
 
-	<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationInbound"
-											  channel="fromChannel"
-											  expect-message="true"
-											  serializer="testSerializer"/>
+	<int-redis:queue-inbound-channel-adapter id="fromChannelEndpoint" queue="si.test.Int3017IntegrationInbound"
+											 channel="fromChannel"
+											 expect-message="true"
+											 auto-startup="false"
+											 serializer="testSerializer"/>
 
 	<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
 
@@ -30,9 +31,11 @@
 		<int-redis:queue-outbound-channel-adapter queue-expression="headers.redis_queue"/>
 	</int:chain>
 
-	<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationSymmetrical"
-											  channel="symmetricalRedisChannel"
-											  serializer=""/>
+	<int-redis:queue-inbound-channel-adapter id="symmetricalRedisChannelEndpoint"
+											 queue="si.test.Int3017IntegrationSymmetrical"
+											 channel="symmetricalRedisChannel"
+											 auto-startup="false"
+											 serializer=""/>
 
 
 	<int:payload-deserializing-transformer input-channel="symmetricalRedisChannel" output-channel="symmetricalOutputChannel"/>

spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2017 the original author or authors.
+ * Copyright 2013-2019 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.
@@ -46,6 +46,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.Lifecycle;
 import org.springframework.data.redis.RedisConnectionFailureException;
 import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -88,9 +89,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;
 
@@ -106,6 +113,7 @@ public void testInt3014Default() throws Exception {
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
 		redisTemplate.afterPropertiesSet();
+		redisTemplate.delete(queueName);
 
 		String payload = "testing";
 
@@ -135,6 +143,7 @@ public void testInt3014Default() throws Exception {
 		assertEquals(payload2, receive.getPayload());
 
 		endpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	@Test
@@ -149,6 +158,7 @@ public void testInt3014ExpectMessageTrue() throws Exception {
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
 		redisTemplate.afterPropertiesSet();
+		redisTemplate.delete(queueName);
 
 		Message<?> message = MessageBuilder.withPayload("testing").build();
 
@@ -187,38 +197,53 @@ public void testInt3014ExpectMessageTrue() throws Exception {
 				Matchers.containsString("java.lang.String cannot be cast to org.springframework.messaging.Message"));
 
 		endpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	@Test
 	@RedisAvailable
 	public void testInt3017IntegrationInbound() throws Exception {
-		String payload = new Date().toString();
-
+		String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
 		RedisTemplate<String, String> 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\":{}}");
 
 		Message<?> receive = this.fromChannel.receive(10000);
 		assertNotNull(receive);
 		assertEquals(payload, receive.getPayload());
+		this.fromChannelEndpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	@Test
 	@RedisAvailable
 	public void testInt3017IntegrationSymmetrical() throws Exception {
+		String queueName = "si.test.Int3017IntegrationSymmetrical";
+		RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
+		redisTemplate.setConnectionFactory(this.connectionFactory);
+		redisTemplate.afterPropertiesSet();
+		redisTemplate.delete(queueName);
+		this.symmetricalRedisChannelEndpoint.start();
 		UUID payload = UUID.randomUUID();
 		Message<UUID> message = MessageBuilder.withPayload(payload)
-				.setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical")
+				.setHeader("redis_queue", queueName)
 				.build();
 
 		this.symmetricalInputChannel.send(message);
 
 		Message<?> receive = this.symmetricalOutputChannel.receive(10000);
 		assertNotNull(receive);
 		assertEquals(payload, receive.getPayload());
+		this.symmetricalRedisChannelEndpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	@Test
@@ -233,6 +258,7 @@ public void testInt3442ProperlyStop() throws Exception {
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
 		redisTemplate.afterPropertiesSet();
+		redisTemplate.delete(queueName);
 
 		while (redisTemplate.boundListOps(queueName).rightPop() != null) {
 			// drain
@@ -277,6 +303,7 @@ public void run() {
 		assertTrue(stopLatch.await(21, TimeUnit.SECONDS));
 
 		Mockito.verify(boundListOperations, atLeastOnce()).rightPush(Mockito.any(byte[].class));
+		redisTemplate.delete(queueName);
 	}
 
 
@@ -292,7 +319,8 @@ public void testInt3196Recovery() throws Exception {
 
 		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(new ApplicationEventPublisher() {
 
@@ -345,6 +373,7 @@ public void publishEvent(Object event) {
 		assertEquals(payload, receive.getPayload());
 
 		endpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	@Test
@@ -359,6 +388,7 @@ public void testInt3932ReadFromLeft() throws Exception {
 		redisTemplate.setKeySerializer(new StringRedisSerializer());
 		redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
 		redisTemplate.afterPropertiesSet();
+		redisTemplate.delete(queueName);
 
 		String payload = "testing";
 
@@ -389,6 +419,7 @@ public void testInt3932ReadFromLeft() throws Exception {
 		assertEquals(payload2, receive.getPayload());
 
 		endpoint.stop();
+		redisTemplate.delete(queueName);
 	}
 
 	private void waitListening(RedisQueueMessageDrivenEndpoint endpoint) throws InterruptedException {