diff --git a/src/main/java/org/springframework/data/redis/stream/RawRedisSerializer.java b/src/main/java/org/springframework/data/redis/serializer/RawRedisSerializer.java
similarity index 84%
rename from src/main/java/org/springframework/data/redis/stream/RawRedisSerializer.java
rename to src/main/java/org/springframework/data/redis/serializer/RawRedisSerializer.java
index 55e60f80c..79597cd44 100644
--- a/src/main/java/org/springframework/data/redis/stream/RawRedisSerializer.java
+++ b/src/main/java/org/springframework/data/redis/serializer/RawRedisSerializer.java
@@ -13,13 +13,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.data.redis.stream;
+package org.springframework.data.redis.serializer;
 
-import org.springframework.data.redis.serializer.RedisSerializer;
-import org.springframework.data.redis.serializer.SerializationException;
 import org.springframework.lang.Nullable;
 
 /**
+ * Raw {@link RedisSerializer} using {@code byte[]}.
+ *
  * @author Mark Paluch
  */
 enum RawRedisSerializer implements RedisSerializer<byte[]> {
diff --git a/src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java b/src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java
index 5ca8d9559..43733835f 100644
--- a/src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java
+++ b/src/main/java/org/springframework/data/redis/serializer/RedisSerializationContext.java
@@ -79,12 +79,23 @@ public interface RedisSerializationContext<K, V> {
 	}
 
 	/**
-	 * Creates a new {@link RedisSerializationContext} using a {@link SerializationPair#raw()} serialization pair.
+	 * Creates a new {@link RedisSerializationContext} using a {@link SerializationPair#raw() ByteBuffer} serialization
+	 * pair.
 	 *
 	 * @return
 	 */
-	static RedisSerializationContext<byte[], byte[]> raw() {
-		return just(SerializationPair.raw());
+	static RedisSerializationContext<ByteBuffer, ByteBuffer> raw() {
+		return just(RedisSerializerToSerializationPairAdapter.raw());
+	}
+
+	/**
+	 * Creates a new {@link RedisSerializationContext} using a {@link RedisSerializer#raw() byte[]} serialization.
+	 *
+	 * @return
+	 * @since 2.2
+	 */
+	static RedisSerializationContext<byte[], byte[]> byteArray() {
+		return just(RedisSerializerToSerializationPairAdapter.byteArray());
 	}
 
 	/**
@@ -207,7 +218,7 @@ public interface RedisSerializationContext<K, V> {
 		 *
 		 * @return a pass through {@link SerializationPair}.
 		 */
-		static <T> SerializationPair<T> raw() {
+		static SerializationPair<ByteBuffer> raw() {
 			return RedisSerializerToSerializationPairAdapter.raw();
 		}
 
diff --git a/src/main/java/org/springframework/data/redis/serializer/RedisSerializer.java b/src/main/java/org/springframework/data/redis/serializer/RedisSerializer.java
index d1d995139..c584a752e 100644
--- a/src/main/java/org/springframework/data/redis/serializer/RedisSerializer.java
+++ b/src/main/java/org/springframework/data/redis/serializer/RedisSerializer.java
@@ -92,6 +92,16 @@ public interface RedisSerializer<T> {
 		return StringRedisSerializer.UTF_8;
 	}
 
+	/**
+	 * Obtain a raw {@link RedisSerializer} that passes thru {@code byte[]}.
+	 *
+	 * @return never {@literal null}.
+	 * @since 2.2
+	 */
+	static RedisSerializer<byte[]> raw() {
+		return RawRedisSerializer.INSTANCE;
+	}
+
 	default boolean canSerialize(Class<?> type) {
 		return ClassUtils.isAssignable(getTargetType(), type);
 	}
diff --git a/src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java b/src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java
index 6a70a182f..6aacba5fd 100644
--- a/src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java
+++ b/src/main/java/org/springframework/data/redis/serializer/RedisSerializerToSerializationPairAdapter.java
@@ -15,6 +15,8 @@
  */
 package org.springframework.data.redis.serializer;
 
+import java.nio.ByteBuffer;
+
 import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -31,18 +33,25 @@ class RedisSerializerToSerializationPairAdapter<T> implements SerializationPair<
 	private final static RedisSerializerToSerializationPairAdapter<?> RAW = new RedisSerializerToSerializationPairAdapter<>(
 			null);
 
+	private final static RedisSerializerToSerializationPairAdapter<byte[]> BYTE_ARRAY = new RedisSerializerToSerializationPairAdapter<>(
+			RedisSerializer.raw());
+
 	private final DefaultSerializationPair pair;
 
-	protected RedisSerializerToSerializationPairAdapter(@Nullable RedisSerializer<T> serializer) {
-		pair = new DefaultSerializationPair(new DefaultRedisElementReader<>(serializer),
+	RedisSerializerToSerializationPairAdapter(@Nullable RedisSerializer<T> serializer) {
+		pair = new DefaultSerializationPair<>(new DefaultRedisElementReader<>(serializer),
 				new DefaultRedisElementWriter<>(serializer));
 	}
 
 	@SuppressWarnings("unchecked")
-	public static <T> SerializationPair<T> raw() {
+	static SerializationPair<ByteBuffer> raw() {
 		return (SerializationPair) RAW;
 	}
 
+	static SerializationPair<byte[]> byteArray() {
+		return BYTE_ARRAY;
+	}
+
 	/**
 	 * Create a {@link SerializationPair} from given {@link RedisSerializer}.
 	 *
diff --git a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
index c17cbde58..8d0aedaca 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
@@ -722,8 +722,8 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 
 			if (this.hashMapper == null) {
 
-				hashKeySerializer(RawRedisSerializer.INSTANCE);
-				hashValueSerializer(RawRedisSerializer.INSTANCE);
+				hashKeySerializer(RedisSerializer.raw());
+				hashValueSerializer(RedisSerializer.raw());
 				return (StreamMessageListenerContainerOptionsBuilder) objectMapper(new ObjectHashMapper());
 			}
 
diff --git a/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java b/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java
index fb8e538ad..c2d9a8ef1 100644
--- a/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java
+++ b/src/test/java/org/springframework/data/redis/core/ReactiveOperationsTestParams.java
@@ -119,7 +119,7 @@ abstract public class ReactiveOperationsTestParams {
 				RedisSerializationContext.<String, Double> newSerializationContext(jdkSerializationRedisSerializer)
 						.key(stringRedisSerializer).value(doubleToStringSerializer).build());
 
-		ReactiveRedisTemplate<byte[], byte[]> rawTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory,
+		ReactiveRedisTemplate<ByteBuffer, ByteBuffer> rawTemplate = new ReactiveRedisTemplate<>(lettuceConnectionFactory,
 				RedisSerializationContext.raw());
 
 		ReactiveRedisTemplate<String, Person> personTemplate = new ReactiveRedisTemplate(lettuceConnectionFactory,
diff --git a/src/test/java/org/springframework/data/redis/serializer/RedisSerializationContextUnitTests.java b/src/test/java/org/springframework/data/redis/serializer/RedisSerializationContextUnitTests.java
index ffb633257..2de1c632a 100644
--- a/src/test/java/org/springframework/data/redis/serializer/RedisSerializationContextUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/serializer/RedisSerializationContextUnitTests.java
@@ -111,6 +111,32 @@ public class RedisSerializationContextUnitTests {
 		assertThat(deserialized).isEqualTo(42);
 	}
 
+	@Test // DATAREDIS-1000
+	public void shouldEncodeAndDecodeRawByteBufferValue() {
+
+		RedisSerializationContext<ByteBuffer, ByteBuffer> serializationContext = RedisSerializationContext
+				.raw();
+
+		ByteBuffer deserialized = serializationContext.getValueSerializationPair()
+				.read(serializationContext.getValueSerializationPair()
+						.write(ByteBuffer.wrap("hello".getBytes())));
+
+		assertThat(deserialized).isEqualTo(ByteBuffer.wrap("hello".getBytes()));
+	}
+
+	@Test // DATAREDIS-1000
+	public void shouldEncodeAndDecodeByteArrayValue() {
+
+		RedisSerializationContext<byte[], byte[]> serializationContext = RedisSerializationContext
+				.byteArray();
+
+		byte[] deserialized = serializationContext.getValueSerializationPair()
+				.read(serializationContext.getValueSerializationPair()
+						.write("hello".getBytes()));
+
+		assertThat(deserialized).isEqualTo("hello".getBytes());
+	}
+
 	private RedisSerializationContext<String, Long> createSerializationContext() {
 
 		return RedisSerializationContext.<String, Long> newSerializationContext() //