Skip to content

Commit 5e1a92d

Browse files
artembilangaryrussell
artembilan
authored and
garyrussell
committed
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**
1 parent 9bdff8c commit 5e1a92d

File tree

2 files changed

+46
-27
lines changed

2 files changed

+46
-27
lines changed

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,17 @@
1212
<util:constant id="redisConnectionFactory"
1313
static-field="org.springframework.integration.redis.rules.RedisAvailableRule.connectionFactory"/>
1414

15+
<util:constant id="TEST_QUEUE"
16+
static-field="org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpointTests.TEST_QUEUE"/>
17+
1518
<int:channel id="fromChannel">
1619
<int:queue/>
1720
</int:channel>
1821

19-
<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationInbound"
22+
<int-redis:queue-inbound-channel-adapter id="fromChannelEndpoint" queue="#{TEST_QUEUE}"
2023
channel="fromChannel"
2124
expect-message="true"
25+
auto-startup="false"
2226
serializer="testSerializer"/>
2327

2428
<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
@@ -28,8 +32,9 @@
2832
<int-redis:queue-outbound-channel-adapter queue-expression="headers.redis_queue"/>
2933
</int:chain>
3034

31-
<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationSymmetrical"
35+
<int-redis:queue-inbound-channel-adapter id="symmetricalRedisChannelEndpoint" queue="#{TEST_QUEUE}"
3236
channel="symmetricalRedisChannel"
37+
auto-startup="false"
3338
serializer=""/>
3439

3540

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

Lines changed: 39 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
import java.util.concurrent.Executors;
3131
import java.util.concurrent.TimeUnit;
3232

33+
import org.junit.After;
34+
import org.junit.Before;
3335
import org.junit.Ignore;
3436
import org.junit.Test;
3537
import org.junit.runner.RunWith;
@@ -41,6 +43,7 @@
4143
import org.springframework.beans.factory.InitializingBean;
4244
import org.springframework.beans.factory.annotation.Autowired;
4345
import org.springframework.context.ApplicationEvent;
46+
import org.springframework.context.Lifecycle;
4447
import org.springframework.data.redis.RedisConnectionFailureException;
4548
import org.springframework.data.redis.RedisSystemException;
4649
import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -78,24 +81,38 @@
7881
@DirtiesContext
7982
public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
8083

84+
public static final String TEST_QUEUE = "testQueue";
85+
8186
@Autowired
8287
private RedisConnectionFactory connectionFactory;
8388

8489
@Autowired
8590
private PollableChannel fromChannel;
8691

92+
@Autowired
93+
private Lifecycle fromChannelEndpoint;
94+
8795
@Autowired
8896
private MessageChannel symmetricalInputChannel;
8997

98+
@Autowired
99+
private Lifecycle symmetricalRedisChannelEndpoint;
100+
90101
@Autowired
91102
private PollableChannel symmetricalOutputChannel;
92103

104+
@Before
105+
@After
106+
public void setUpTearDown() {
107+
RedisTemplate<String, ?> redisTemplate = new RedisTemplate<>();
108+
redisTemplate.setConnectionFactory(this.connectionFactory);
109+
redisTemplate.delete(TEST_QUEUE);
110+
}
111+
93112
@Test
94113
@RedisAvailable
95114
@SuppressWarnings("unchecked")
96115
public void testInt3014Default() {
97-
String queueName = "si.test.redisQueueInboundChannelAdapterTests";
98-
99116
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
100117
redisTemplate.setConnectionFactory(this.connectionFactory);
101118
redisTemplate.setEnableDefaultSerializer(false);
@@ -105,16 +122,16 @@ public void testInt3014Default() {
105122

106123
String payload = "testing";
107124

108-
redisTemplate.boundListOps(queueName).leftPush(payload);
125+
redisTemplate.boundListOps(TEST_QUEUE).leftPush(payload);
109126

110127
Date payload2 = new Date();
111128

112-
redisTemplate.boundListOps(queueName).leftPush(payload2);
129+
redisTemplate.boundListOps(TEST_QUEUE).leftPush(payload2);
113130

114131
PollableChannel channel = new QueueChannel();
115132

116133
RedisQueueMessageDrivenEndpoint endpoint =
117-
new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
134+
new RedisQueueMessageDrivenEndpoint(TEST_QUEUE, this.connectionFactory);
118135
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
119136
endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
120137
endpoint.setOutputChannel(channel);
@@ -137,8 +154,6 @@ public void testInt3014Default() {
137154
@RedisAvailable
138155
@SuppressWarnings("unchecked")
139156
public void testInt3014ExpectMessageTrue() {
140-
String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
141-
142157
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
143158
redisTemplate.setConnectionFactory(this.connectionFactory);
144159
redisTemplate.setEnableDefaultSerializer(false);
@@ -148,16 +163,16 @@ public void testInt3014ExpectMessageTrue() {
148163

149164
Message<?> message = MessageBuilder.withPayload("testing").build();
150165

151-
redisTemplate.boundListOps(queueName).leftPush(message);
166+
redisTemplate.boundListOps(TEST_QUEUE).leftPush(message);
152167

153-
redisTemplate.boundListOps(queueName).leftPush("test");
168+
redisTemplate.boundListOps(TEST_QUEUE).leftPush("test");
154169

155170
PollableChannel channel = new QueueChannel();
156171

157172
PollableChannel errorChannel = new QueueChannel();
158173

159174
RedisQueueMessageDrivenEndpoint endpoint =
160-
new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
175+
new RedisQueueMessageDrivenEndpoint(TEST_QUEUE, this.connectionFactory);
161176
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
162177
endpoint.setExpectMessage(true);
163178
endpoint.setOutputChannel(channel);
@@ -186,53 +201,55 @@ public void testInt3014ExpectMessageTrue() {
186201
@Test
187202
@RedisAvailable
188203
public void testInt3017IntegrationInbound() {
204+
this.fromChannelEndpoint.start();
189205
String payload = new Date().toString();
190206

191207
RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
192208
redisTemplate.setConnectionFactory(this.connectionFactory);
193209
redisTemplate.afterPropertiesSet();
194210

195-
redisTemplate.boundListOps("si.test.Int3017IntegrationInbound")
211+
redisTemplate.boundListOps(TEST_QUEUE)
196212
.leftPush("{\"payload\":\"" + payload + "\",\"headers\":{}}");
197213

198214
Message<?> receive = this.fromChannel.receive(10000);
199215
assertThat(receive).isNotNull();
200216
assertThat(receive.getPayload()).isEqualTo(payload);
217+
this.fromChannelEndpoint.stop();
201218
}
202219

203220
@Test
204221
@RedisAvailable
205222
public void testInt3017IntegrationSymmetrical() {
223+
this.symmetricalRedisChannelEndpoint.start();
206224
UUID payload = UUID.randomUUID();
207225
Message<UUID> message = MessageBuilder.withPayload(payload)
208-
.setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical")
226+
.setHeader("redis_queue", TEST_QUEUE)
209227
.build();
210228

211229
this.symmetricalInputChannel.send(message);
212230

213231
Message<?> receive = this.symmetricalOutputChannel.receive(10000);
214232
assertThat(receive).isNotNull();
215233
assertThat(receive.getPayload()).isEqualTo(payload);
234+
this.symmetricalRedisChannelEndpoint.stop();
216235
}
217236

218237
@Test
219238
@RedisAvailable
220239
@SuppressWarnings("unchecked")
221240
public void testInt3442ProperlyStop() throws Exception {
222-
String queueName = "si.test.testInt3442ProperlyStopTest";
223-
224241
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
225242
redisTemplate.setConnectionFactory(this.connectionFactory);
226243
redisTemplate.setEnableDefaultSerializer(false);
227244
redisTemplate.setKeySerializer(new StringRedisSerializer());
228245
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
229246
redisTemplate.afterPropertiesSet();
230247

231-
while (redisTemplate.boundListOps(queueName).rightPop() != null) {
248+
while (redisTemplate.boundListOps(TEST_QUEUE).rightPop() != null) {
232249
// drain
233250
}
234251

235-
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName,
252+
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(TEST_QUEUE,
236253
this.connectionFactory);
237254
BoundListOperations<String, byte[]> boundListOperations =
238255
TestUtils.getPropertyValue(endpoint, "boundListOperations", BoundListOperations.class);
@@ -252,7 +269,7 @@ public void testInt3442ProperlyStop() throws Exception {
252269
waitListening(endpoint);
253270
dfa.setPropertyValue("listening", false);
254271

255-
redisTemplate.boundListOps(queueName).leftPush("foo");
272+
redisTemplate.boundListOps(TEST_QUEUE).leftPush("foo");
256273

257274
CountDownLatch stopLatch = new CountDownLatch(1);
258275

@@ -271,14 +288,13 @@ public void testInt3442ProperlyStop() throws Exception {
271288
@RedisAvailable
272289
@Ignore("LettuceConnectionFactory doesn't support proper reinitialization after 'destroy()'")
273290
public void testInt3196Recovery() throws Exception {
274-
String queueName = "test.si.Int3196Recovery";
275291
QueueChannel channel = new QueueChannel();
276292

277293
final List<ApplicationEvent> exceptionEvents = new ArrayList<>();
278294

279295
final CountDownLatch exceptionsLatch = new CountDownLatch(2);
280296

281-
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName,
297+
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(TEST_QUEUE,
282298
this.connectionFactory);
283299
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
284300
endpoint.setApplicationEventPublisher(event -> {
@@ -315,7 +331,7 @@ public void testInt3196Recovery() throws Exception {
315331

316332
String payload = "testing";
317333

318-
redisTemplate.boundListOps(queueName).leftPush(payload);
334+
redisTemplate.boundListOps(TEST_QUEUE).leftPush(payload);
319335

320336
Message<?> receive = channel.receive(10000);
321337
assertThat(receive).isNotNull();
@@ -328,8 +344,6 @@ public void testInt3196Recovery() throws Exception {
328344
@RedisAvailable
329345
@SuppressWarnings("unchecked")
330346
public void testInt3932ReadFromLeft() {
331-
String queueName = "si.test.redisQueueInboundChannelAdapterTests3932";
332-
333347
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
334348
redisTemplate.setConnectionFactory(this.connectionFactory);
335349
redisTemplate.setEnableDefaultSerializer(false);
@@ -339,16 +353,16 @@ public void testInt3932ReadFromLeft() {
339353

340354
String payload = "testing";
341355

342-
redisTemplate.boundListOps(queueName).rightPush(payload);
356+
redisTemplate.boundListOps(TEST_QUEUE).rightPush(payload);
343357

344358
Date payload2 = new Date();
345359

346-
redisTemplate.boundListOps(queueName).rightPush(payload2);
360+
redisTemplate.boundListOps(TEST_QUEUE).rightPush(payload2);
347361

348362
PollableChannel channel = new QueueChannel();
349363

350364
RedisQueueMessageDrivenEndpoint endpoint =
351-
new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
365+
new RedisQueueMessageDrivenEndpoint(TEST_QUEUE, this.connectionFactory);
352366
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
353367
endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
354368
endpoint.setOutputChannel(channel);

0 commit comments

Comments
 (0)