DATAREDIS-1128 - Improve test stability.

Replace condition-based polling/waiting with Awaitility. Improve assertions for timing-based checks. Increase number of HyperLogLog samples to reduce probability of clashing elements.

Remove long-waiting sleeps that did not lead to an assertion.

Original Pull Request: #525

src/test/java/org/springframework/data/redis/SpinBarrier.java

@@ -15,10 +15,17 @@
  */
 package org.springframework.data.redis;
 
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+
 /**
  * @author Jennifer Hickey
+ * @author Mark Paluch
+ * @deprecated Use {@link Awaitility}.
  */
-abstract public class SpinBarrier {
+@Deprecated
+public abstract class SpinBarrier {
 
 	/**
 	 * Periodically tests for a condition until it is met or a timeout occurs
@@ -28,16 +35,12 @@ abstract public class SpinBarrier {
 	 * @return true if condition passes, false if condition does not pass within timeout
 	 */
 	public static boolean waitFor(TestCondition condition, long timeout) {
-		boolean passes = false;
-		for (long currentTime = System.currentTimeMillis(); System.currentTimeMillis() - currentTime < timeout;) {
-			if (condition.passes()) {
-				passes = true;
-				break;
-			}
-			try {
-				Thread.sleep(100);
-			} catch (InterruptedException e) {}
+
+		try {
+			Awaitility.await().atMost(timeout, TimeUnit.MILLISECONDS).until(condition::passes);
+			return true;
+		} catch (Exception e) {
+			return false;
 		}
-		return passes;
 	}
 }

src/test/java/org/springframework/data/redis/config/NamespaceTest.java

@@ -17,8 +17,6 @@ package org.springframework.data.redis.config;
 
 import static org.assertj.core.api.Assertions.*;
 
-import java.util.concurrent.TimeUnit;
-
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -26,12 +24,12 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner;
-import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.annotation.ProfileValueSourceConfiguration;
 import org.springframework.test.context.ContextConfiguration;
 
 /**
  * @author Costin Leau
+ * @author Mark Paluch
  */
 @RunWith(RelaxedJUnit4ClassRunner.class)
 @ContextConfiguration("namespace.xml")
@@ -45,24 +43,13 @@ public class NamespaceTest {
 	@Autowired private StubErrorHandler handler;
 
 	@Test
-	@IfProfileValue(name = "runLongTests", value = "true")
 	public void testSanityTest() throws Exception {
 		assertThat(container.isRunning()).isTrue();
-		Thread.sleep(TimeUnit.SECONDS.toMillis(8));
 	}
 
 	@Test
-	@IfProfileValue(name = "runLongTests", value = "true")
 	public void testWithMessages() throws Exception {
 		template.convertAndSend("x1", "[X]test");
 		template.convertAndSend("z1", "[Z]test");
-		Thread.sleep(TimeUnit.SECONDS.toMillis(8));
-	}
-
-	public void testErrorHandler() throws Exception {
-		int index = handler.throwables.size();
-		template.convertAndSend("exception", "test1");
-		handler.throwables.pollLast(3, TimeUnit.SECONDS);
-		assertThat(handler.throwables.size()).isEqualTo(index + 1);
 	}
 }

src/test/java/org/springframework/data/redis/connection/AbstractConnectionIntegrationTests.java

@@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.assertj.core.data.Offset;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.AssumptionViolatedException;
@@ -396,7 +397,8 @@ public abstract class AbstractConnectionIntegrationTests {
 		Thread.sleep(200);
 		connection.scriptKill();
 		getResults();
-		assertThat(waitFor(() -> scriptDead.get(), 3000l)).isTrue();
+
+		Awaitility.await().untilTrue(scriptDead);
 	}
 
 	@Test
@@ -833,7 +835,7 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		connection.watch("testitnow".getBytes());
 		// Give some time for watch to be asynch executed in extending tests
-		Thread.sleep(500);
+		Thread.sleep(200);
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection());
 		conn2.set("testitnow", "something");
 		conn2.close();
@@ -852,11 +854,10 @@ public abstract class AbstractConnectionIntegrationTests {
 
 		connection.watch("testitnow".getBytes());
 		connection.unwatch();
-
 		connection.multi();
 
 		// Give some time for unwatch to be asynch executed
-		Thread.sleep(100);
+		Thread.sleep(200);
 		DefaultStringRedisConnection conn2 = new DefaultStringRedisConnection(connectionFactory.getConnection());
 		conn2.set("testitnow", "something");
 

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceClusterConnectionTests.java

@@ -2365,7 +2365,7 @@ public class LettuceClusterConnectionTests implements ClusterConnectionTests {
 		nativeConnection.set(KEY_1, VALUE_1);
 		nativeConnection.get(KEY_1);
 
-		assertThat(clusterConnection.keyCommands().idletime(KEY_1_BYTES)).isEqualTo(Duration.ofSeconds(0));
+		assertThat(clusterConnection.keyCommands().idletime(KEY_1_BYTES)).isLessThanOrEqualTo(Duration.ofSeconds(2));
 	}
 
 	@Test // DATAREDIS-716

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

@@ -315,25 +315,6 @@ public class LettuceConnectionFactoryTests {
 		pool.destroy();
 	}
 
-	@Ignore("Uncomment this test to manually check connection reuse in a pool scenario")
-	@Test
-	public void testLotsOfConnections() throws InterruptedException {
-		// Running a netstat here should show only the 8 conns from the pool (plus 2 from setUp and 1 from factory2
-		// afterPropertiesSet for shared conn)
-		DefaultLettucePool pool = new DefaultLettucePool(SettingsUtils.getHost(), SettingsUtils.getPort());
-		pool.afterPropertiesSet();
-		final LettuceConnectionFactory factory2 = new LettuceConnectionFactory(pool);
-		factory2.afterPropertiesSet();
-
-		ConnectionFactoryTracker.add(factory2);
-
-		for (int i = 1; i < 1000; i++) {
-			Thread th = new Thread(() -> factory2.getConnection().bRPop(50000, "foo".getBytes()));
-			th.start();
-		}
-		Thread.sleep(234234234);
-	}
-
 	@Ignore("Redis must have requirepass set to run this test")
 	@Test
 	public void testConnectWithPassword() {

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionIntegrationTests.java

@@ -27,6 +27,7 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.awaitility.Awaitility;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -275,7 +276,8 @@ public class LettuceConnectionIntegrationTests extends AbstractConnectionIntegra
 		th.start();
 		Thread.sleep(1000);
 		connection.scriptKill();
-		assertThat(waitFor(scriptDead::get, 3000l)).isTrue();
+
+		Awaitility.await().untilTrue(scriptDead);
 	}
 
 	@Test

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionPipelineIntegrationTests.java

@@ -22,6 +22,7 @@ import static org.springframework.data.redis.SpinBarrier.*;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.awaitility.Awaitility;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -79,7 +80,8 @@ public class LettuceConnectionPipelineIntegrationTests extends AbstractConnectio
 		Thread.sleep(1000);
 		connection.scriptKill();
 		getResults();
-		assertThat(waitFor(scriptDead::get, 3000l)).isTrue();
+
+		Awaitility.await().untilTrue(scriptDead);
 	}
 
 	@Test

src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveScriptingCommandsTests.java

@@ -20,6 +20,7 @@ import static org.junit.Assume.*;
 import static org.springframework.data.redis.SpinBarrier.*;
 
 import io.lettuce.core.ScriptOutputType;
+import org.awaitility.Awaitility;
 import reactor.test.StepVerifier;
 
 import java.nio.ByteBuffer;
@@ -174,7 +175,7 @@ public class LettuceReactiveScriptingCommandsTests extends LettuceReactiveComman
 
 		connection.scriptingCommands().scriptKill().as(StepVerifier::create).expectNext("OK").verifyComplete();
 
-		assertThat(waitFor(scriptDead::get, 3000L)).isTrue();
+		Awaitility.await().untilTrue(scriptDead);
 	}
 
 	private static ByteBuffer wrap(String content) {

src/test/java/org/springframework/data/redis/core/DefaultHyperLogLogOperationsTests.java

@@ -35,6 +35,7 @@ import org.springframework.test.annotation.IfProfileValue;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @RunWith(Parameterized.class)
 @IfProfileValue(name = "redisVersion", value = "2.8.9+")
@@ -132,10 +133,11 @@ public class DefaultHyperLogLogOperationsTests<K, V> {
 
 		K key2 = keyFactory.instance();
 		V v4 = valueFactory.instance();
+		V v5 = valueFactory.instance();
 
 		hyperLogLogOps.add(key, v1, v2, v3);
-		hyperLogLogOps.add(key2, v4);
-		assertThat(hyperLogLogOps.size(key, key2)).isEqualTo(4L);
+		hyperLogLogOps.add(key2, v4, v5);
+		assertThat(hyperLogLogOps.size(key, key2)).isGreaterThan(3L);
 	}
 
 	@Test // DATAREDIS-308
@@ -149,16 +151,14 @@ public class DefaultHyperLogLogOperationsTests<K, V> {
 
 		K sourceKey_2 = keyFactory.instance();
 		V v4 = valueFactory.instance();
+		V v5 = valueFactory.instance();
 
 		K desinationKey = keyFactory.instance();
 
 		hyperLogLogOps.add(sourceKey_1, v1, v2, v3);
-		hyperLogLogOps.add(sourceKey_2, v4);
-
-		Thread.sleep(10); // give redis a little time to catch up
+		hyperLogLogOps.add(sourceKey_2, v4, v5);
 		hyperLogLogOps.union(desinationKey, sourceKey_1, sourceKey_2);
-		Thread.sleep(10); // give redis a little time to catch up
 
-		assertThat(hyperLogLogOps.size(desinationKey)).isEqualTo(4L);
+		assertThat(hyperLogLogOps.size(desinationKey)).isGreaterThan(3L);
 	}
 }

src/test/java/org/springframework/data/redis/listener/KeyExpirationEventMessageListenerTests.java

@@ -19,7 +19,9 @@ import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -37,6 +39,7 @@ import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 
 /**
  * @author Christoph Strobl
+ * @author Mark Paluch
  */
 @RunWith(MockitoJUnitRunner.class)
 public class KeyExpirationEventMessageListenerTests {
@@ -62,18 +65,15 @@ public class KeyExpirationEventMessageListenerTests {
 		listener = new KeyExpirationEventMessageListener(container);
 		listener.setApplicationEventPublisher(publisherMock);
 		listener.init();
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+			connection.flushAll();
+		}
 	}
 
 	@After
 	public void tearDown() throws Exception {
 
-		RedisConnection connection = connectionFactory.getConnection();
-		try {
-			connection.flushAll();
-		} finally {
-			connection.close();
-		}
-
 		listener.destroy();
 		container.destroy();
 		if (connectionFactory instanceof DisposableBean) {
@@ -82,25 +82,23 @@ public class KeyExpirationEventMessageListenerTests {
 	}
 
 	@Test // DATAREDIS-425
-	public void listenerShouldPublishEventCorrectly() throws InterruptedException {
+	public void listenerShouldPublishEventCorrectly() {
 
 		byte[] key = ("to-expire:" + UUID.randomUUID().toString()).getBytes();
+		AtomicBoolean called = new AtomicBoolean();
+		doAnswer(invocation -> {
+			called.set(true);
+			return null;
+		}).when(publisherMock).publishEvent(any(ApplicationEvent.class));
 
-		RedisConnection connection = connectionFactory.getConnection();
-		try {
-			connection.setEx(key, 2, "foo".getBytes());
+		try (RedisConnection connection = connectionFactory.getConnection()) {
 
-			int iteration = 0;
-			while (connection.get(key) != null || iteration >= 3) {
-
-				Thread.sleep(2000);
-				iteration++;
-			}
-		} finally {
-			connection.close();
+			connection.pSetEx(key, 1, key);
+			Awaitility.await().until(() -> !connection.exists(key));
 		}
 
-		Thread.sleep(2000);
+		Awaitility.await().untilTrue(called);
+
 		ArgumentCaptor<ApplicationEvent> captor = ArgumentCaptor.forClass(ApplicationEvent.class);
 
 		verify(publisherMock, times(1)).publishEvent(captor.capture());
@@ -112,18 +110,13 @@ public class KeyExpirationEventMessageListenerTests {
 
 		byte[] key = ("to-delete:" + UUID.randomUUID().toString()).getBytes();
 
-		RedisConnection connection = connectionFactory.getConnection();
-		try {
+		try (RedisConnection connection = connectionFactory.getConnection()) {
 
 			connection.setEx(key, 10, "foo".getBytes());
-			Thread.sleep(2000);
 			connection.del(key);
-			Thread.sleep(2000);
-		} finally {
-			connection.close();
 		}
 
-		Thread.sleep(2000);
-		verifyZeroInteractions(publisherMock);
+		Thread.sleep(500);
+		verifyNoInteractions(publisherMock);
 	}
 }

src/test/java/org/springframework/data/redis/listener/PubSubAwaitUtil.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2020 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.data.redis.listener;
+
+import io.lettuce.core.codec.ByteArrayCodec;
+import io.lettuce.core.output.ArrayOutput;
+import io.lettuce.core.output.IntegerOutput;
+
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import org.awaitility.Awaitility;
+
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnection;
+
+/**
+ * Utility providing await support.
+ * 
+ * @author Mark Paluch
+ */
+class PubSubAwaitUtil {
+
+	/**
+	 * Run the {@link Runnable} and wait until the number of Pub/Sub patterns has increased in comparison to before
+	 * running the callback.
+	 * 
+	 * @param connectionFactory
+	 * @param runnable
+	 */
+	public static void runAndAwaitPatternSubscription(RedisConnectionFactory connectionFactory, Runnable runnable) {
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+
+			Number before = numPat(connection);
+
+			runnable.run();
+
+			runAndAwait(runnable, () -> {
+
+				Number after = numPat(connection);
+
+				return after.longValue() > before.longValue();
+			});
+		}
+	}
+
+	/**
+	 * Run the {@link Runnable} and wait until the number of channel subscribers has increased in comparison to before
+	 * running the callback.
+	 * 
+	 * @param connectionFactory
+	 * @param channel
+	 * @param runnable
+	 */
+	public static void runAndAwaitChannelSubscription(RedisConnectionFactory connectionFactory, String channel,
+			Runnable runnable) {
+
+		try (RedisConnection connection = connectionFactory.getConnection()) {
+
+			Number before = numSub(connection, channel);
+
+			runnable.run();
+
+			runAndAwait(runnable, () -> {
+
+				Number after = numSub(connection, channel);
+
+				return after.longValue() > before.longValue();
+			});
+		}
+	}
+
+	private static long numPat(RedisConnection connection) {
+
+		if (connection instanceof LettuceConnection) {
+			return ((Number) ((LettuceConnection) connection).execute("PUBSUB", new IntegerOutput<>(ByteArrayCodec.INSTANCE),
+					"NUMPAT".getBytes())).longValue();
+		}
+
+		return ((Number) connection.execute("PUBSUB", "NUMPAT".getBytes())).longValue();
+	}
+
+	private static long numSub(RedisConnection connection, String channel) {
+
+		List<?> pubsub;
+		if (connection instanceof LettuceConnection) {
+			pubsub = (List<?>) ((LettuceConnection) connection).execute("PUBSUB", new ArrayOutput<>(ByteArrayCodec.INSTANCE),
+					"NUMSUB".getBytes(), channel.getBytes());
+		} else {
+			pubsub = (List<?>) connection.execute("PUBSUB", "NUMSUB".getBytes(), channel.getBytes());
+		}
+
+		return ((Number) pubsub.get(1)).longValue();
+	}
+
+	private static void runAndAwait(Runnable runnable, Callable<Boolean> predicate) {
+
+		runnable.run();
+		Awaitility.await().until(predicate);
+	}
+}

src/test/java/org/springframework/data/redis/listener/PubSubResubscribeTests.java

@@ -17,7 +17,6 @@ package org.springframework.data.redis.listener;
 
 import static org.assertj.core.api.Assertions.*;
 import static org.junit.Assume.*;
-import static org.springframework.data.redis.SpinBarrier.*;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -26,6 +25,7 @@ import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
@@ -39,7 +39,6 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
-import org.junit.runners.model.Statement;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
@@ -150,17 +149,15 @@ public class PubSubResubscribeTests {
 		container = new RedisMessageListenerContainer();
 		container.setConnectionFactory(template.getConnectionFactory());
 		container.setBeanName("container");
-		container.addMessageListener(adapter, new ChannelTopic(CHANNEL));
 		container.setTaskExecutor(new SyncTaskExecutor());
 		container.setSubscriptionExecutor(new SimpleAsyncTaskExecutor());
 		container.afterPropertiesSet();
 		container.start();
-
-		waitFor(container.getConnectionFactory().getConnection()::isSubscribed, 1000);
 	}
 
 	@After
 	public void tearDown() {
+		container.stop();
 		bag.clear();
 	}
 
@@ -180,7 +177,6 @@ public class PubSubResubscribeTests {
 
 		// remove adapter from all channels
 		container.addMessageListener(anotherListener, new PatternTopic(PATTERN));
-		container.removeMessageListener(adapter);
 
 		// Wait for async subscription tasks to setup
 		Thread.sleep(400);
@@ -275,18 +271,24 @@ public class PubSubResubscribeTests {
 	@Test
 	public void testInitializeContainerWithMultipleTopicsIncludingPattern() throws Exception {
 
-		container.removeMessageListener(adapter);
+		assumeFalse(isClusterAware(template.getConnectionFactory()));
+
 		container.stop();
-		container.addMessageListener(adapter,
-				Arrays.asList(new Topic[] { new ChannelTopic(CHANNEL), new PatternTopic("s*") }));
-		container.start();
+
+		String uniqueChannel = "random-" + UUID.randomUUID().toString();
+		PubSubAwaitUtil.runAndAwaitChannelSubscription(template.getConnectionFactory(), uniqueChannel, () -> {
+
+			container.addMessageListener(adapter,
+					Arrays.asList(new Topic[] { new ChannelTopic(uniqueChannel), new PatternTopic("s*") }));
+			container.start();
+		});
 
 		// timing: There's currently no other way to synchronize
 		// than to hope the subscribe/unsubscribe are executed within the time.
-		Thread.sleep(1000);
+		Thread.sleep(50);
 
 		template.convertAndSend("somechannel", "HELLO");
-		template.convertAndSend(CHANNEL, "WORLD");
+		template.convertAndSend(uniqueChannel, "WORLD");
 
 		Set<String> set = new LinkedHashSet<>();
 		set.add(bag.poll(500, TimeUnit.MILLISECONDS));
@@ -313,17 +315,16 @@ public class PubSubResubscribeTests {
 	}
 
 	private static boolean clusterAvailable() {
+		return new RedisClusterRule().isAvailable();
+	}
 
-		try {
-			new RedisClusterRule().apply(new Statement() {
-				@Override
-				public void evaluate() throws Throwable {
+	private static boolean isClusterAware(RedisConnectionFactory connectionFactory) {
 
-				}
-			}, null).evaluate();
-		} catch (Throwable throwable) {
-			return false;
+		if (connectionFactory instanceof LettuceConnectionFactory) {
+			return ((LettuceConnectionFactory) connectionFactory).isClusterAware();
+		} else if (connectionFactory instanceof JedisConnectionFactory) {
+			return ((JedisConnectionFactory) connectionFactory).isRedisClusterAware();
 		}
-		return true;
+		return false;
 	}
 }

src/test/java/org/springframework/data/redis/listener/PubSubTests.java

@@ -20,6 +20,7 @@ import static org.junit.Assume.*;
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.concurrent.BlockingDeque;
@@ -39,6 +40,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.ObjectFactory;
+import org.springframework.data.redis.connection.ConnectionUtils;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
@@ -185,13 +187,15 @@ public class PubSubTests<T> {
 	@Test // DATAREDIS-251
 	public void testStartListenersToNoSpecificChannelTest() throws InterruptedException {
 
-		assumeTrue(isClusterAware(template.getConnectionFactory()));
+		assumeFalse(isClusterAware(template.getConnectionFactory()));
+		assumeTrue(ConnectionUtils.isJedis(template.getConnectionFactory()));
 
-		container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
-		container.addMessageListener(adapter, Arrays.asList(new PatternTopic("*")));
-		container.start();
+		PubSubAwaitUtil.runAndAwaitPatternSubscription(template.getRequiredConnectionFactory(), () -> {
 
-		Thread.sleep(1000); // give the container a little time to recover
+			container.removeMessageListener(adapter, new ChannelTopic(CHANNEL));
+			container.addMessageListener(adapter, Collections.singletonList(new PatternTopic(CHANNEL + "*")));
+			container.start();
+		});
 
 		T payload = getT();
 

src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerIntegrationTests.java

@@ -27,6 +27,7 @@ import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -34,6 +35,7 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
+
 import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.connection.ReactiveSubscription;
 import org.springframework.data.redis.connection.ReactiveSubscription.ChannelMessage;
@@ -198,16 +200,7 @@ public class ReactiveRedisMessageListenerContainerIntegrationTests {
 	private static Runnable awaitSubscription(Supplier<Collection<ReactiveSubscription>> activeSubscriptions) {
 
 		return () -> {
-
-			try {
-
-				while (activeSubscriptions.get().isEmpty()) {
-					Thread.sleep(10);
-				}
-
-			} catch (InterruptedException e) {
-				return;
-			}
+			Awaitility.await().until(() -> !activeSubscriptions.get().isEmpty());
 		};
 	}
 }

src/test/java/org/springframework/data/redis/stream/StreamMessageListenerContainerIntegrationTests.java

@@ -30,6 +30,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import org.awaitility.Awaitility;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -66,6 +67,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 	private static final RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(
 			SettingsUtils.getHost(), SettingsUtils.getPort());
+	public static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(2);
 
 	private static RedisConnectionFactory connectionFactory;
 
@@ -115,7 +117,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
@@ -143,7 +145,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add(ObjectRecord.create("my-stream", "value1"));
 
@@ -167,7 +169,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add(ObjectRecord.create("my-stream", new LoginEvent("Walter", "White")));
 
@@ -192,7 +194,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"),
 				StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
 
@@ -218,7 +220,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		Subscription subscription = container.receiveAutoAck(Consumer.from("my-group", "my-consumer"),
 				StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
 
@@ -245,7 +247,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"),
 				StreamOffset.create("my-stream", ReadOffset.lastConsumed()), it -> {});
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		Throwable error = failures.poll(1, TimeUnit.SECONDS);
 		assertThat(failures).isEmpty();
@@ -271,14 +273,11 @@ public class StreamMessageListenerContainerIntegrationTests {
 
 		container.start();
 		Subscription subscription = container.register(readRequest, it -> {});
-
 		subscription.await(Duration.ofSeconds(1));
 
 		redisTemplate.delete("my-stream");
 
-		subscription.await(Duration.ofSeconds(1));
-
-		assertThat(failures.poll(1, TimeUnit.SECONDS)).isNotNull();
+		assertThat(failures.poll(3, TimeUnit.SECONDS)).isNotNull();
 		assertThat(subscription.isActive()).isFalse();
 
 		cancelAwait(subscription);
@@ -305,7 +304,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.register(readRequest, it -> {});
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.delete("my-stream");
 
@@ -326,7 +325,7 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 		cancelAwait(subscription);
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value4"));
@@ -344,17 +343,15 @@ public class StreamMessageListenerContainerIntegrationTests {
 		container.start();
 		Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		container.stop();
 
-		while (subscription.isActive()) {
-			Thread.sleep(10);
-		}
+		Awaitility.await().atMost(DEFAULT_TIMEOUT).until(() -> !subscription.isActive());
 
 		container.start();
 
-		subscription.await(Duration.ofSeconds(2));
+		subscription.await(DEFAULT_TIMEOUT);
 
 		redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
 
@@ -363,13 +360,11 @@ public class StreamMessageListenerContainerIntegrationTests {
 		cancelAwait(subscription);
 	}
 
-	private static void cancelAwait(Subscription subscription) throws InterruptedException {
+	private static void cancelAwait(Subscription subscription) {
 
 		subscription.cancel();
 
-		while (subscription.isActive()) {
-			Thread.sleep(10);
-		}
+		Awaitility.await().atMost(DEFAULT_TIMEOUT).until(() -> !subscription.isActive());
 	}
 
 	private int getNumberOfPending(String stream, String group) {

src/test/java/org/springframework/data/redis/stream/StreamReceiverIntegrationTests.java

@@ -183,24 +183,18 @@ public class StreamReceiverIntegrationTests {
 
 		messages.as(publisher -> StepVerifier.create(publisher, 0)) //
 				.thenRequest(1) //
+				.thenAwait(Duration.ofMillis(500)) //
 				.then(() -> {
-					try {
-						Thread.sleep(500);
-						reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"))
-								.subscribe();
-					} catch (InterruptedException e) {}
+					reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1")).subscribe();
 				}) //
 				.expectNextCount(1) //
 				.then(() -> {
 					reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")).subscribe();
 				}) //
 				.thenRequest(1) //
+				.thenAwait(Duration.ofMillis(500)) //
 				.then(() -> {
-					try {
-						Thread.sleep(500);
-						reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3"))
-								.subscribe();
-					} catch (InterruptedException e) {}
+					reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3")).subscribe();
 				}).consumeNextWith(it -> {
 
 					assertThat(it.getStream()).isEqualTo("my-stream");

src/test/java/org/springframework/data/redis/test/util/RedisClusterRule.java

@@ -30,6 +30,7 @@ import org.springframework.data.redis.connection.jedis.JedisConverters;
  * Simple {@link TestRule} implementation that check Redis is running in cluster mode.
  *
  * @author Christoph Strobl
+ * @author Mark Paluch
  * @since 1.7
  */
 public class RedisClusterRule extends ExternalResource {
@@ -67,6 +68,13 @@ public class RedisClusterRule extends ExternalResource {
 		return this.clusterConfig;
 	}
 
+	/**
+	 * @return {@literal true} if Redis Cluster is available.
+	 */
+	public boolean isAvailable() {
+		return "cluster".equals(mode);
+	}
+
 	private void init() {
 
 		if (clusterConfig == null) {