diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java
index 9656b9ff7..14a7931ad 100644
--- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java
+++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamMessageListenerContainer.java
@@ -89,13 +89,17 @@ class DefaultStreamMessageListenerContainer<K, V extends Record<K, ?>> implement
 
 	private static StreamReadOptions getStreamReadOptions(StreamMessageListenerContainerOptions<?, ?> options) {
 
-		StreamReadOptions streamReadOptions = StreamReadOptions.empty().count(options.getBatchSize());
+		StreamReadOptions readOptions = StreamReadOptions.empty();
 
-		if (!options.getPollTimeout().isZero()) {
-			streamReadOptions = streamReadOptions.block(options.getPollTimeout());
+		if (options.getBatchSize().isPresent()) {
+			readOptions = readOptions.count(options.getBatchSize().getAsInt());
 		}
 
-		return streamReadOptions;
+		if (!options.getPollTimeout().isZero()) {
+			readOptions = readOptions.block(options.getPollTimeout());
+		}
+
+		return readOptions;
 	}
 
 	private RedisTemplate<K, V> createRedisTemplate(RedisConnectionFactory connectionFactory,
diff --git a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
index 5ec5d9afd..8de6a2c22 100644
--- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
+++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
@@ -73,7 +73,11 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 				.key(options.getKeySerializer()).hashKey(options.getHashKeySerializer())
 				.hashValue(options.getHashValueSerializer()).build();
 
-		StreamReadOptions readOptions = StreamReadOptions.empty().count(options.getBatchSize());
+		StreamReadOptions readOptions = StreamReadOptions.empty();
+
+		if (options.getBatchSize().isPresent()) {
+			readOptions = readOptions.count(options.getBatchSize().getAsInt());
+		}
 		if (!options.getPollTimeout().isZero()) {
 			readOptions = readOptions.block(options.getPollTimeout());
 		}
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 1f969a359..de7aa8706 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
@@ -16,6 +16,7 @@
 package org.springframework.data.redis.stream;
 
 import java.time.Duration;
+import java.util.OptionalInt;
 import java.util.concurrent.Executor;
 import java.util.function.Predicate;
 
@@ -474,7 +475,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 	class StreamMessageListenerContainerOptions<K, V extends Record<K, ?>> {
 
 		private final Duration pollTimeout;
-		private final int batchSize;
+		private final @Nullable Integer batchSize;
 		private final RedisSerializer<K> keySerializer;
 		private final RedisSerializer<Object> hashKeySerializer;
 		private final RedisSerializer<Object> hashValueSerializer;
@@ -484,7 +485,8 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 		private final Executor executor;
 
 		@SuppressWarnings("unchecked")
-		private StreamMessageListenerContainerOptions(Duration pollTimeout, int batchSize, RedisSerializer<K> keySerializer,
+		private StreamMessageListenerContainerOptions(Duration pollTimeout, @Nullable Integer batchSize,
+				RedisSerializer<K> keySerializer,
 				RedisSerializer<Object> hashKeySerializer, RedisSerializer<Object> hashValueSerializer,
 				@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper, ErrorHandler errorHandler,
 				Executor executor) {
@@ -518,10 +520,10 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 		/**
 		 * Batch size polling using the {@code COUNT} option during reads.
 		 *
-		 * @return
+		 * @return the batch size.
 		 */
-		public int getBatchSize() {
-			return batchSize;
+		public OptionalInt getBatchSize() {
+			return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty();
 		}
 
 		public RedisSerializer<K> getKeySerializer() {
@@ -575,7 +577,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 	class StreamMessageListenerContainerOptionsBuilder<K, V extends Record<K, ?>> {
 
 		private Duration pollTimeout = Duration.ofSeconds(2);
-		private int batchSize = 1;
+		private @Nullable Integer batchSize;
 		private RedisSerializer<K> keySerializer;
 		private RedisSerializer<Object> hashKeySerializer;
 		private RedisSerializer<Object> hashValueSerializer;
diff --git a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
index 9c04df042..de196aa08 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
@@ -19,6 +19,7 @@ import reactor.core.publisher.Flux;
 
 import java.nio.ByteBuffer;
 import java.time.Duration;
+import java.util.OptionalInt;
 
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
 import org.springframework.data.redis.connection.stream.Consumer;
@@ -186,14 +187,14 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 	class StreamReceiverOptions<K, V extends Record<K, ?>> {
 
 		private final Duration pollTimeout;
-		private final int batchSize;
+		private final @Nullable Integer batchSize;
 		private final SerializationPair<K> keySerializer;
 		private final SerializationPair<Object> hashKeySerializer;
 		private final SerializationPair<Object> hashValueSerializer;
 		private final @Nullable Class<Object> targetType;
 		private final @Nullable HashMapper<Object, Object, Object> hashMapper;
 
-		private StreamReceiverOptions(Duration pollTimeout, int batchSize, SerializationPair<K> keySerializer,
+		private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair<K> keySerializer,
 				SerializationPair<Object> hashKeySerializer, SerializationPair<Object> hashValueSerializer,
 				@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper) {
 
@@ -240,10 +241,10 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		/**
 		 * Batch size polling using the {@code COUNT} option during reads.
 		 *
-		 * @return
+		 * @return the batch size.
 		 */
-		public int getBatchSize() {
-			return batchSize;
+		public OptionalInt getBatchSize() {
+			return batchSize != null ? OptionalInt.of(batchSize) : OptionalInt.empty();
 		}
 
 		public SerializationPair<K> getKeySerializer() {
@@ -281,7 +282,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 	class StreamReceiverOptionsBuilder<K, V extends Record<K, ?>> {
 
 		private Duration pollTimeout = Duration.ofSeconds(2);
-		private int batchSize = 1;
+		private @Nullable Integer batchSize;
 		private SerializationPair<K> keySerializer;
 		private SerializationPair<Object> hashKeySerializer;
 		private SerializationPair<Object> hashValueSerializer;