diff --git a/pom.xml b/pom.xml
index a67dc48660..9d9624b813 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-redis</artifactId>
-	<version>3.2.0-SNAPSHOT</version>
+	<version>3.2.0-GH-2633-SNAPSHOT</version>
 
 	<name>Spring Data Redis</name>
 	<description>Spring Data module for Redis</description>
diff --git a/src/main/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBean.java b/src/main/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBean.java
index 3c0f8ca537..376181c1fd 100644
--- a/src/main/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBean.java
+++ b/src/main/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBean.java
@@ -15,10 +15,13 @@
  */
 package org.springframework.data.redis.support.collections;
 
+import java.util.function.Supplier;
+
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.SmartFactoryBean;
 import org.springframework.data.redis.connection.DataType;
+import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.util.Lazy;
 import org.springframework.lang.Nullable;
@@ -27,11 +30,13 @@
 
 /**
  * Factory bean that facilitates creation of Redis-based collections. Supports list, set, zset (or sortedSet), map (or
- * hash) and properties. Will use the key type if it exists or to create a dedicated collection (Properties vs Map).
- * Otherwise uses the provided type (default is list).
+ * hash) and properties. Uses the key and {@link CollectionType} to determine what collection type to use. The factory
+ * verifies the key type if a {@link CollectionType} is specified. Defaults to {@link CollectionType#LIST}.
  *
  * @author Costin Leau
  * @author Christoph Strobl
+ * @author Mark Paluch
+ * @see RedisStore
  */
 public class RedisCollectionFactoryBean implements SmartFactoryBean<RedisStore>, BeanNameAware, InitializingBean {
 
@@ -39,48 +44,74 @@ public class RedisCollectionFactoryBean implements SmartFactoryBean<RedisStore>,
 	 * Collection types supported by this factory.
 	 *
 	 * @author Costin Leau
+	 * @author Mark Paluch
 	 */
 	public enum CollectionType {
 		LIST {
 
+			@Override
 			public DataType dataType() {
 				return DataType.LIST;
 			}
 		},
 		SET {
 
+			@Override
 			public DataType dataType() {
 				return DataType.SET;
 			}
 		},
 		ZSET {
 
+			@Override
 			public DataType dataType() {
 				return DataType.ZSET;
 			}
 		},
 		MAP {
 
+			@Override
 			public DataType dataType() {
 				return DataType.HASH;
 			}
 		},
 		PROPERTIES {
 
+			@Override
 			public DataType dataType() {
 				return DataType.HASH;
 			}
 		};
 
 		abstract DataType dataType();
+
+		/**
+		 * Attempt to find a {@link CollectionType} by {@link DataType}. Defaults to {@link Supplier ifNotFound} when
+		 * {@code dataType} is {@literal null} or the collection type cannot be determined.
+		 *
+		 * @param dataType the {@link DataType} to look up.
+		 * @param ifNotFound supplier for a default value.
+		 * @since 3.2
+		 */
+		static CollectionType findCollectionType(@Nullable DataType dataType, Supplier<CollectionType> ifNotFound) {
+
+			for (CollectionType collectionType : values()) {
+				if (collectionType.dataType() == dataType) {
+					return collectionType;
+				}
+			}
+
+			return ifNotFound.get();
+		}
 	}
 
-	private @Nullable Lazy<RedisStore> store;
-	private @Nullable CollectionType type = null;
+	private @Nullable CollectionType type;
 	private @Nullable RedisTemplate<String, ?> template;
 	private @Nullable String key;
 	private @Nullable String beanName;
 
+	private @Nullable Lazy<RedisStore> store;
+
 	@Override
 	public void afterPropertiesSet() {
 
@@ -93,46 +124,40 @@ public void afterPropertiesSet() {
 
 		store = Lazy.of(() -> {
 
-			DataType dt = template.type(key);
+			DataType keyType = template.type(key);
 
 			// can't create store
-			Assert.isTrue(!DataType.STRING.equals(dt), "Cannot create store on keys of type 'string'");
+			Assert.isTrue(!DataType.STREAM.equals(keyType), "Cannot create store on keys of type 'STREAM'");
 
-			RedisStore tmp = createStore(dt);
+			if (this.type == null) {
+				this.type = CollectionType.findCollectionType(keyType, () -> CollectionType.LIST);
+			}
 
-			if (tmp == null) {
-				if (type == null) {
-					type = CollectionType.LIST;
-				}
-				tmp = createStore(type.dataType());
+			if (keyType != null && DataType.NONE != keyType && this.type.dataType() != keyType) {
+				throw new IllegalArgumentException(
+						String.format("Cannot create collection type '%s' for a key containing '%s'", this.type, keyType));
 			}
-			return tmp;
+
+			return createStore(this.type, key, template);
 		});
 	}
 
-	@SuppressWarnings("unchecked")
-	private RedisStore createStore(DataType dt) {
-		switch (dt) {
-			case LIST:
-				return RedisList.create(key, template);
-
-			case SET:
-				return new DefaultRedisSet(key, template);
-
-			case ZSET:
-				return RedisZSet.create(key, template);
+	private RedisStore createStore(CollectionType collectionType, String key, RedisOperations<String, ?> operations) {
 
-			case HASH:
-				if (CollectionType.PROPERTIES.equals(type)) {
-					return new RedisProperties(key, template);
-				}
-				return new DefaultRedisMap(key, template);
-		}
-		return null;
+		return switch (collectionType) {
+			case LIST -> RedisList.create(key, operations);
+			case SET -> new DefaultRedisSet<>(key, operations);
+			case ZSET -> RedisZSet.create(key, operations);
+			case PROPERTIES -> new RedisProperties(key, operations);
+			case MAP -> new DefaultRedisMap<>(key, operations);
+		};
 	}
 
 	@Override
 	public RedisStore getObject() {
+
+		Assert.state(store != null,
+				"RedisCollectionFactoryBean is not initialized. Ensure to initialize this factory by calling afterPropertiesSet() before obtaining the factory object.");
 		return store.get();
 	}
 
diff --git a/src/test/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBeanTests.java b/src/test/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBeanTests.java
index 1ef51d63b2..3af400c75f 100644
--- a/src/test/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBeanTests.java
+++ b/src/test/java/org/springframework/data/redis/support/collections/RedisCollectionFactoryBeanTests.java
@@ -17,9 +17,11 @@
 
 import static org.assertj.core.api.Assertions.*;
 
+import java.util.Map;
+
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-
 import org.springframework.data.redis.ObjectFactory;
 import org.springframework.data.redis.StringObjectFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
@@ -30,7 +32,10 @@
 import org.springframework.data.redis.test.extension.RedisStanalone;
 
 /**
+ * Integration tests for {@link RedisCollectionFactoryBean}.
+ *
  * @author Costin Leau
+ * @author Mark Paluch
  */
 public class RedisCollectionFactoryBeanTests {
 
@@ -45,6 +50,12 @@ public class RedisCollectionFactoryBeanTests {
 		this.template = new StringRedisTemplate(jedisConnFactory);
 	}
 
+	@BeforeEach
+	void setUp() {
+		this.template.delete("key");
+		this.template.delete("nosrt");
+	}
+
 	@AfterEach
 	void tearDown() throws Exception {
 		// clean up the whole db
@@ -86,8 +97,30 @@ void testNone() throws Exception {
 		assertThat(store).isInstanceOf(DefaultRedisList.class);
 	}
 
+	@Test // GH-2633
+	void testExisting() {
+
+		template.delete("key");
+		template.opsForHash().put("key", "k", "v");
+
+		assertThat(createCollection("key")).isInstanceOf(DefaultRedisMap.class);
+		assertThat(createCollection("key", CollectionType.MAP)).isInstanceOf(DefaultRedisMap.class);
+
+		template.delete("key");
+		template.opsForSet().add("key", "1", "2");
+
+		assertThat(createCollection("key")).isInstanceOf(DefaultRedisSet.class);
+		assertThat(createCollection("key", CollectionType.SET)).isInstanceOf(DefaultRedisSet.class);
+
+		template.delete("key");
+		template.opsForList().leftPush("key", "1", "2");
+
+		assertThat(createCollection("key")).isInstanceOf(DefaultRedisList.class);
+		assertThat(createCollection("key", CollectionType.LIST)).isInstanceOf(DefaultRedisList.class);
+	}
+
 	@Test
-	void testExistingCol() throws Exception {
+	void testExistingCol() {
 		String key = "set";
 		String val = "value";
 
@@ -102,6 +135,28 @@ void testExistingCol() throws Exception {
 
 		col = createCollection(key, CollectionType.PROPERTIES);
 		assertThat(col).isInstanceOf(RedisProperties.class);
+	}
+
+	@Test // GH-2633
+	void testIncompatibleCollections() {
+
+		template.opsForValue().set("key", "value");
+		assertThatIllegalArgumentException().isThrownBy(() -> createCollection("key", CollectionType.LIST))
+				.withMessageContaining("Cannot create collection type 'LIST' for a key containing 'STRING'");
+
+		template.delete("key");
+		template.opsForList().leftPush("key", "value");
+		assertThatIllegalArgumentException().isThrownBy(() -> createCollection("key", CollectionType.SET))
+				.withMessageContaining("Cannot create collection type 'SET' for a key containing 'LIST'");
+
+	}
+
+	@Test // GH-2633
+	void shouldFailForStreamCreation() {
+
+		template.opsForStream().add("key", Map.of("k", "v"));
+		assertThatIllegalArgumentException().isThrownBy(() -> createCollection("key", CollectionType.LIST))
+				.withMessageContaining("Cannot create store on keys of type 'STREAM'");
 
 	}
 }