From cc6bde665ad561a8295e5bce6a8e94387b2a0678 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Tue, 16 Feb 2016 11:20:55 +0100 Subject: [PATCH] DATAREDIS-463 - Improve test synchronization. Several tests rely heavily on time to perform synchronization. Therfore we remove Thread.sleep from tests which do not require a sleep at all, use a condition where possible and increase timeouts on tests known to fail very likely due to a short Thread.sleep. Original Pull Request: #171 --- .../AbstractConnectionIntegrationTests.java | 44 ++++++++++------ .../LettuceClusterConnectionTests.java | 2 +- .../data/redis/core/RedisTemplateTests.java | 2 +- .../listener/PubSubResubscribeTests.java | 52 ++++++++++++++----- .../listener/SubscriptionConnectionTests.java | 11 ++-- 5 files changed, 74 insertions(+), 37 deletions(-) diff --git a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java index 2d72eebab..513281a53 100644 --- a/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java @@ -35,6 +35,7 @@ import java.util.Properties; import java.util.Set; import java.util.UUID; import java.util.concurrent.BlockingDeque; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -80,6 +81,7 @@ import org.springframework.test.annotation.ProfileValueSourceConfiguration; * @author Jennifer Hickey * @author Christoph Strobl * @author Thomas Darimont + * @author Mark Paluch */ @ProfileValueSourceConfiguration(RedisTestProfileValueSource.class) public abstract class AbstractConnectionIntegrationTests { @@ -343,10 +345,12 @@ public abstract class AbstractConnectionIntegrationTests { assumeTrue(RedisVersionUtils.atLeast("2.6", byteConnection)); initConnection(); final AtomicBoolean scriptDead = new AtomicBoolean(false); + final CountDownLatch sync = new CountDownLatch(1); Thread th = new Thread(new Runnable() { public void run() { DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection()); try { + sync.countDown(); conn2.eval("local time=1 while time < 10000000000 do time=time+1 end", ReturnType.BOOLEAN, 0); } catch (DataAccessException e) { scriptDead.set(true); @@ -355,7 +359,8 @@ public abstract class AbstractConnectionIntegrationTests { } }); th.start(); - Thread.sleep(1000); + sync.await(2, TimeUnit.SECONDS); + Thread.sleep(200); connection.scriptKill(); getResults(); assertTrue(waitFor(new TestCondition() { @@ -380,11 +385,10 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testPersist() throws Exception { connection.set("exp3", "true"); - actual.add(connection.expire("exp3", 1)); + actual.add(connection.expire("exp3", 30)); actual.add(connection.persist("exp3")); - Thread.sleep(1500); - actual.add(connection.exists("exp3")); - verifyResults(Arrays.asList(new Object[] { true, true, true })); + actual.add(connection.ttl("exp3")); + verifyResults(Arrays.asList(new Object[] { true, true, -1L })); } @Test @@ -414,7 +418,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBRPopTimeout() throws Exception { actual.add(connection.bRPop(1, "alist")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -422,7 +425,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBLPopTimeout() throws Exception { actual.add(connection.bLPop(1, "alist")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -430,7 +432,6 @@ public abstract class AbstractConnectionIntegrationTests { @IfProfileValue(name = "runLongTests", value = "true") public void testBRPopLPushTimeout() throws Exception { actual.add(connection.bRPopLPush(1, "alist", "foo")); - Thread.sleep(1500l); verifyResults(Arrays.asList(new Object[] { null })); } @@ -633,12 +634,16 @@ public abstract class AbstractConnectionIntegrationTests { Thread th = new Thread(new Runnable() { public void run() { - // sleep 1/2 second to let the registration happen + // sync to let the registration happen + waitFor(new TestCondition() { + @Override + public boolean passes() { + return connection.isSubscribed(); + } + }, 2000); try { Thread.sleep(500); - } catch (InterruptedException ex) { - throw new RuntimeException(ex); - } + } catch (InterruptedException o_O) {} // open a new connection RedisConnection connection2 = connectionFactory.getConnection(); @@ -681,12 +686,17 @@ public abstract class AbstractConnectionIntegrationTests { Thread th = new Thread(new Runnable() { public void run() { - // sleep 1/2 second to let the registration happen + // sync to let the registration happen + waitFor(new TestCondition() { + @Override + public boolean passes() { + return connection.isSubscribed(); + } + }, 2000); + try { Thread.sleep(500); - } catch (InterruptedException ex) { - throw new RuntimeException(ex); - } + } catch (InterruptedException o_O) {} // open a new connection RedisConnection connection2 = connectionFactory.getConnection(); @@ -1007,7 +1017,7 @@ public abstract class AbstractConnectionIntegrationTests { actual.add(connection.get("testing")); connection.restore("testing".getBytes(), 100l, (byte[]) results.get(0)); verifyResults(Arrays.asList(new Object[] { 1l, null })); - assertTrue(waitFor(new KeyExpired("testing"), 300l)); + assertTrue(waitFor(new KeyExpired("testing"), 400l)); } @Test diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java index 8bc60a2f3..6cdc3e60d 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java @@ -86,7 +86,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests { public void setUp() { client = new RedisClusterClient(Builder.redis(CLUSTER_HOST, MASTER_NODE_1_PORT) - .withTimeout(100, TimeUnit.MILLISECONDS).build()); + .withTimeout(500, TimeUnit.MILLISECONDS).build()); nativeConnection = client.connectCluster(); clusterConnection = new LettuceClusterConnection(client); } diff --git a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java index f6df2210f..5df2daa3d 100644 --- a/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java +++ b/src/test/java/org/springframework/data/redis/core/RedisTemplateTests.java @@ -443,7 +443,7 @@ public class RedisTemplateTests { final K key1 = keyFactory.instance(); V value1 = valueFactory.instance(); redisTemplate.boundValueOps(key1).set(value1); - redisTemplate.expire(key1, 250, TimeUnit.MILLISECONDS); + redisTemplate.expire(key1, 500, TimeUnit.MILLISECONDS); assertTrue(redisTemplate.getExpire(key1, TimeUnit.MILLISECONDS) > 0l); // Timeout is longer because expire will be 1 sec if pExpire not supported diff --git a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java index 077b2398f..79d50ccdd 100644 --- a/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java +++ b/src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2014 the original author or authors. + * Copyright 2011-2016 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. @@ -19,6 +19,7 @@ package org.springframework.data.redis.listener; import static org.hamcrest.core.Is.*; import static org.junit.Assert.*; import static org.junit.Assume.*; +import static org.springframework.data.redis.SpinBarrier.*; import java.util.ArrayList; import java.util.Arrays; @@ -44,6 +45,7 @@ import org.springframework.core.task.SyncTaskExecutor; import org.springframework.data.redis.ConnectionFactoryTracker; import org.springframework.data.redis.RedisTestProfileValueSource; import org.springframework.data.redis.SettingsUtils; +import org.springframework.data.redis.TestCondition; import org.springframework.data.redis.connection.ConnectionUtils; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; @@ -58,6 +60,7 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; * @author Costin Leau * @author Jennifer Hickey * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) public class PubSubResubscribeTests { @@ -149,7 +152,12 @@ public class PubSubResubscribeTests { container.afterPropertiesSet(); container.start(); - Thread.sleep(1000); + waitFor(new TestCondition() { + @Override + public boolean passes() { + return container.getConnectionFactory().getConnection().isSubscribed(); + } + }, 1000); } @After @@ -175,14 +183,17 @@ public class PubSubResubscribeTests { container.addMessageListener(anotherListener, new PatternTopic(PATTERN)); container.removeMessageListener(adapter); + // Wait for async subscription tasks to setup + Thread.sleep(400); + // test no messages are sent just to patterns template.convertAndSend(CHANNEL, payload1); template.convertAndSend(ANOTHER_CHANNEL, payload2); // anotherListener receives both messages List msgs = new ArrayList(); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); assertEquals(2, msgs.size()); assertTrue(msgs.contains(payload1)); @@ -190,21 +201,29 @@ public class PubSubResubscribeTests { msgs.clear(); // unsubscribed adapter did not receive message - assertNull(bag.poll(1, TimeUnit.SECONDS)); + assertNull(bag.poll(500, TimeUnit.MILLISECONDS)); // bind original listener on another channel container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); + // Wait for async subscription tasks to setup + Thread.sleep(400); + template.convertAndSend(CHANNEL, payload1); template.convertAndSend(ANOTHER_CHANNEL, payload2); // original listener received only one message on another channel - assertEquals(payload2, bag.poll(1, TimeUnit.SECONDS)); - assertNull(bag.poll(1, TimeUnit.SECONDS)); + msgs.clear(); + msgs.add(bag.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag.poll(500, TimeUnit.MILLISECONDS)); + + assertTrue(msgs.contains(payload2)); + assertTrue(msgs.contains(null)); // another listener receives messages on both channels - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); - msgs.add(bag2.poll(1, TimeUnit.SECONDS)); + msgs.clear(); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); + msgs.add(bag2.poll(500, TimeUnit.MILLISECONDS)); assertEquals(2, msgs.size()); assertTrue(msgs.contains(payload1)); assertTrue(msgs.contains(payload2)); @@ -225,6 +244,10 @@ public class PubSubResubscribeTests { container.addMessageListener(adapter, new ChannelTopic(ANOTHER_CHANNEL)); container.removeMessageListener(null, new ChannelTopic(CHANNEL)); + // timing: There's currently no other way to synchronize + // than to hope the subscribe/unsubscribe are executed within the time. + Thread.sleep(400); + // Listener removed from channel template.convertAndSend(CHANNEL, payload1); template.convertAndSend(CHANNEL, payload2); @@ -234,8 +257,8 @@ public class PubSubResubscribeTests { template.convertAndSend(ANOTHER_CHANNEL, anotherPayload2); Set set = new LinkedHashSet(); - set.add(bag.poll(1, TimeUnit.SECONDS)); - set.add(bag.poll(1, TimeUnit.SECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); assertFalse(set.contains(payload1)); assertFalse(set.contains(payload2)); @@ -259,15 +282,16 @@ public class PubSubResubscribeTests { Arrays.asList(new Topic[] { new ChannelTopic(CHANNEL), new PatternTopic("s*") })); container.start(); - // Wait for async subscription tasks to setup + // timing: There's currently no other way to synchronize + // than to hope the subscribe/unsubscribe are executed within the time. Thread.sleep(1000); template.convertAndSend("somechannel", "HELLO"); template.convertAndSend(CHANNEL, "WORLD"); Set set = new LinkedHashSet(); - set.add(bag.poll(1, TimeUnit.SECONDS)); - set.add(bag.poll(1, TimeUnit.SECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); + set.add(bag.poll(500, TimeUnit.MILLISECONDS)); assertEquals(new HashSet(Arrays.asList(new String[] { "HELLO", "WORLD" })), set); } diff --git a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java index 2435dd549..80325c481 100644 --- a/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java +++ b/src/test/java/org/springframework/data/redis/listener/SubscriptionConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2011-2014 the original author or authors. + * Copyright 2011-2016 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. @@ -47,6 +47,7 @@ import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; * @author Jennifer Hickey * @author Thomas Darimont * @author Christoph Strobl + * @author Mark Paluch */ @RunWith(Parameterized.class) public class SubscriptionConnectionTests { @@ -135,9 +136,11 @@ public class SubscriptionConnectionTests { container.afterPropertiesSet(); container.start(); - // Need to sleep shortly as jedis cannot deal propery with multiple repsonses within one connection - // @see https://github.com/xetorthio/jedis/issues/186 - Thread.sleep(1000); + if (connectionFactory instanceof JedisConnectionFactory) { + // Need to sleep shortly as jedis cannot deal propery with multiple repsonses within one connection + // @see https://github.com/xetorthio/jedis/issues/186 + Thread.sleep(100); + } container.stop(); containers.add(container);