diff --git a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java
index fc0703061..335159d50 100644
--- a/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/ReactiveStreamCommands.java
@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.connection;
 
-import org.springframework.data.redis.connection.stream.StreamRecords;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -31,13 +30,14 @@ import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.CommandResponse;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.KeyCommand;
 import org.springframework.data.redis.connection.ReactiveRedisConnection.NumericResponse;
+import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
 import org.springframework.data.redis.connection.stream.ByteBufferRecord;
 import org.springframework.data.redis.connection.stream.Consumer;
 import org.springframework.data.redis.connection.stream.ReadOffset;
 import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.connection.stream.StreamOffset;
 import org.springframework.data.redis.connection.stream.StreamReadOptions;
-import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
+import org.springframework.data.redis.connection.stream.StreamRecords;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -139,7 +139,7 @@ public interface ReactiveStreamCommands {
 	 * @param key the stream key.
 	 * @param group name of the consumer group.
 	 * @param recordIds record Id's to acknowledge.
-	 * @return
+	 * @return {@link Mono} emitting the nr of acknowledged messages.
 	 * @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
 	 */
 	default Mono<Long> xAck(ByteBuffer key, String group, String... recordIds) {
@@ -157,7 +157,7 @@ public interface ReactiveStreamCommands {
 	 * @param key the stream key.
 	 * @param group name of the consumer group.
 	 * @param recordIds record Id's to acknowledge.
-	 * @return
+	 * @return {@link Mono} emitting the nr of acknowledged messages.
 	 * @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
 	 */
 	default Mono<Long> xAck(ByteBuffer key, String group, RecordId... recordIds) {
@@ -173,7 +173,7 @@ public interface ReactiveStreamCommands {
 	 * Acknowledge one or more records as processed.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting the nr of acknowledged messages per {@link AcknowledgeCommand}.
 	 * @see <a href="http://redis.io/commands/xack">Redis Documentation: XACK</a>
 	 */
 	Flux<NumericResponse<AcknowledgeCommand, Long>> xAck(Publisher<AcknowledgeCommand> commands);
@@ -230,7 +230,7 @@ public interface ReactiveStreamCommands {
 		}
 
 		/**
-		 * @return
+		 * @return the actual {@link ByteBufferRecord#getValue()}
 		 */
 		public Map<ByteBuffer, ByteBuffer> getBody() {
 			return record.getValue();
@@ -246,7 +246,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key must not be {@literal null}.
 	 * @param body must not be {@literal null}.
-	 * @return
+	 * @return {@link Mono} emitting the server generated {@link RecordId id}.
 	 * @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
 	 */
 	default Mono<RecordId> xAdd(ByteBuffer key, Map<ByteBuffer, ByteBuffer> body) {
@@ -261,7 +261,7 @@ public interface ReactiveStreamCommands {
 	 * Add stream record with given {@literal body} to {@literal key}.
 	 *
 	 * @param record must not be {@literal null}.
-	 * @return
+	 * @return {@link Mono} the {@link RecordId id}.
 	 * @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
 	 */
 	default Mono<RecordId> xAdd(ByteBufferRecord record) {
@@ -275,7 +275,7 @@ public interface ReactiveStreamCommands {
 	 * Add stream record with given {@literal body} to {@literal key}.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting the {@link RecordId} on by for for the given {@link AddStreamRecord} commands.
 	 * @see <a href="http://redis.io/commands/xadd">Redis Documentation: XADD</a>
 	 */
 	Flux<CommandResponse<AddStreamRecord, RecordId>> xAdd(Publisher<AddStreamRecord> commands);
@@ -349,7 +349,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key the stream key.
 	 * @param recordIds stream record Id's.
-	 * @return number of removed entries.
+	 * @return {@link Mono} emitting the number of removed entries.
 	 * @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
 	 */
 	default Mono<Long> xDel(ByteBuffer key, String... recordIds) {
@@ -366,7 +366,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key the stream key.
 	 * @param recordIds stream record Id's.
-	 * @return number of removed entries.
+	 * @return {@link Mono} emitting the number of removed entries.
 	 * @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
 	 */
 	default Mono<Long> xDel(ByteBuffer key, RecordId... recordIds) {
@@ -382,7 +382,7 @@ public interface ReactiveStreamCommands {
 	 * number of IDs passed in case certain IDs do not exist.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Mono} emitting the number of removed entries.
 	 * @see <a href="http://redis.io/commands/xdel">Redis Documentation: XDEL</a>
 	 */
 	Flux<CommandResponse<DeleteCommand, Long>> xDel(Publisher<DeleteCommand> commands);
@@ -391,7 +391,7 @@ public interface ReactiveStreamCommands {
 	 * Get the size of the stream stored at {@literal key}.
 	 *
 	 * @param key must not be {@literal null}.
-	 * @return length of the stream.
+	 * @return {@link Mono} emitting the length of the stream.
 	 * @see <a href="http://redis.io/commands/xlen">Redis Documentation: XLEN</a>
 	 */
 	default Mono<Long> xLen(ByteBuffer key) {
@@ -405,7 +405,7 @@ public interface ReactiveStreamCommands {
 	 * Get the size of the stream stored at {@link KeyCommand#getKey()}
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting the length of the stream per {@link KeyCommand}.
 	 * @see <a href="http://redis.io/commands/xlen">Redis Documentation: XLEN</a>
 	 */
 	Flux<NumericResponse<KeyCommand, Long>> xLen(Publisher<KeyCommand> commands);
@@ -501,7 +501,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key the stream key.
 	 * @param range must not be {@literal null}.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting with members of the stream.
 	 * @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
 	 */
 	default Flux<ByteBufferRecord> xRange(ByteBuffer key, Range<String> range) {
@@ -514,7 +514,7 @@ public interface ReactiveStreamCommands {
 	 * @param key the stream key.
 	 * @param range must not be {@literal null}.
 	 * @param limit must not be {@literal null}.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting with members of the stream.
 	 * @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
 	 */
 	default Flux<ByteBufferRecord> xRange(ByteBuffer key, Range<String> range, Limit limit) {
@@ -531,7 +531,7 @@ public interface ReactiveStreamCommands {
 	 * Read records from a stream within a specific {@link Range} applying a {@link Limit}.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting with members of the stream per {@link RangeCommand}.
 	 * @see <a href="http://redis.io/commands/xrange">Redis Documentation: XRANGE</a>
 	 */
 	Flux<CommandResponse<RangeCommand, Flux<ByteBufferRecord>>> xRange(Publisher<RangeCommand> commands);
@@ -633,7 +633,7 @@ public interface ReactiveStreamCommands {
 	 * Read records from one or more {@link StreamOffset}s.
 	 *
 	 * @param streams the streams to read from.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting with members of the stream.
 	 * @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
 	 */
 	default Flux<ByteBufferRecord> xRead(StreamOffset<ByteBuffer>... streams) {
@@ -645,7 +645,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param readOptions read arguments.
 	 * @param streams the streams to read from.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting with members of the stream.
 	 * @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
 	 */
 	default Flux<ByteBufferRecord> xRead(StreamReadOptions readOptions, StreamOffset<ByteBuffer>... streams) {
@@ -661,7 +661,7 @@ public interface ReactiveStreamCommands {
 	 * Read records from one or more {@link StreamOffset}s.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting the members of the stream per {@link ReadCommand}.
 	 * @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
 	 * @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
 	 */
@@ -675,7 +675,7 @@ public interface ReactiveStreamCommands {
 		private final @Nullable ReadOffset offset;
 
 		public GroupCommand(@Nullable ByteBuffer key, GroupCommandAction action, @Nullable String groupName,
-				@Nullable String consumerName, ReadOffset offset) {
+				@Nullable String consumerName, @Nullable ReadOffset offset) {
 
 			super(key);
 			this.action = action;
@@ -732,7 +732,7 @@ public interface ReactiveStreamCommands {
 		}
 
 		public enum GroupCommandAction {
-			CREATE, SET_ID, DESTROY, DELETE_CONSUMER;
+			CREATE, SET_ID, DESTROY, DELETE_CONSUMER
 		}
 	}
 
@@ -800,7 +800,7 @@ public interface ReactiveStreamCommands {
 	 * Execute the given {@link GroupCommand} to {@literal create, destroy,... } groups.
 	 *
 	 * @param commands
-	 * @return
+	 * @return {@link Flux} emitting the results of the {@link GroupCommand} one by one.
 	 */
 	Flux<CommandResponse<GroupCommand, String>> xGroup(Publisher<GroupCommand> commands);
 
@@ -809,7 +809,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param consumer consumer/group.
 	 * @param streams the streams to read from.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting the members of the stream
 	 * @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
 	 */
 	default Flux<ByteBufferRecord> xReadGroup(Consumer consumer, StreamOffset<ByteBuffer>... streams) {
@@ -822,7 +822,7 @@ public interface ReactiveStreamCommands {
 	 * @param consumer consumer/group.
 	 * @param readOptions read arguments.
 	 * @param streams the streams to read from.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting the members of the stream.
 	 * @see <a href="http://redis.io/commands/xreadgroup">Redis Documentation: XREADGROUP</a>
 	 */
 	default Flux<ByteBufferRecord> xReadGroup(Consumer consumer, StreamReadOptions readOptions,
@@ -841,7 +841,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key the stream key.
 	 * @param range must not be {@literal null}.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting the members of the stream in reverse.
 	 * @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
 	 */
 	default Flux<ByteBufferRecord> xRevRange(ByteBuffer key, Range<String> range) {
@@ -854,7 +854,7 @@ public interface ReactiveStreamCommands {
 	 * @param key the stream key.
 	 * @param range must not be {@literal null}.
 	 * @param limit must not be {@literal null}.
-	 * @return list with members of the resulting stream.
+	 * @return {@link Flux} emitting the members of the stream in reverse.
 	 * @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
 	 */
 	default Flux<ByteBufferRecord> xRevRange(ByteBuffer key, Range<String> range, Limit limit) {
@@ -871,7 +871,7 @@ public interface ReactiveStreamCommands {
 	 * Read records from a stream within a specific {@link Range} applying a {@link Limit} in reverse order.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting the members of the stream in reverse.
 	 * @see <a href="http://redis.io/commands/xrevrange">Redis Documentation: XREVRANGE</a>
 	 */
 	Flux<CommandResponse<RangeCommand, Flux<ByteBufferRecord>>> xRevRange(Publisher<RangeCommand> commands);
@@ -929,7 +929,7 @@ public interface ReactiveStreamCommands {
 	 *
 	 * @param key the stream key.
 	 * @param count length of the stream.
-	 * @return number of removed entries.
+	 * @return {@link Mono} emitting the number of removed entries.
 	 * @see <a href="http://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
 	 */
 	default Mono<Long> xTrim(ByteBuffer key, long count) {
@@ -943,7 +943,7 @@ public interface ReactiveStreamCommands {
 	 * Trims the stream to {@code count} elements.
 	 *
 	 * @param commands must not be {@literal null}.
-	 * @return
+	 * @return {@link Flux} emitting the number of removed entries per {@link TrimCommand}.
 	 * @see <a href="http://redis.io/commands/xtrim">Redis Documentation: XTRIM</a>
 	 */
 	Flux<NumericResponse<KeyCommand, Long>> xTrim(Publisher<TrimCommand> commands);
diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
index 1373a3650..aed0a3580 100644
--- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
+++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
@@ -120,6 +120,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
 	}
 
 	@Override
+	@SuppressWarnings({"unchecked", "rawtypes"})
 	public Flux<CommandResponse<GroupCommand, String>> xGroup(Publisher<GroupCommand> commands) {
 
 		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
diff --git a/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java b/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java
index f0b2d6455..ea233fb9e 100644
--- a/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java
+++ b/src/main/java/org/springframework/data/redis/connection/stream/StreamRecords.java
@@ -94,7 +94,7 @@ public class StreamRecords {
 	/**
 	 * Obtain new instance of {@link RecordBuilder} to fluently create {@link Record records}.
 	 *
-	 * @return
+	 * @return new instance of {@link RecordBuilder}.
 	 */
 	public static RecordBuilder<?> newRecord() {
 		return new RecordBuilder<>(null, RecordId.autoGenerate());
diff --git a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java
index e9468ea5d..1fdab753c 100644
--- a/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/DefaultStreamOperations.java
@@ -123,6 +123,7 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
 	 */
 	@Nullable
 	@Override
+	@SuppressWarnings("unchecked")
 	public RecordId add(Record<K, ?> record) {
 
 		Assert.notNull(record, "Record must not be null");
@@ -278,11 +279,7 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
 		return hashValueSerializer() != null;
 	}
 
-	protected HV deserializeHashValue(byte[] bytes, Class<HV> targetType) {
-		return hashValueSerializerPresent() ? (HV) hashValueSerializer().deserialize(bytes)
-				: objectMapper.getConversionService().convert(bytes, targetType);
-	}
-
+	@SuppressWarnings({"rawtypes", "unchecked"})
 	private byte[] serialize(Object value, RedisSerializer serializer) {
 
 		Object _value = value;
@@ -302,9 +299,13 @@ class DefaultStreamOperations<K, HK, HV> extends AbstractOperations<K, Object> i
 
 	abstract class RecordDeserializingRedisCallback implements RedisCallback<List<MapRecord<K, HK, HV>>> {
 
+		@SuppressWarnings("unchecked")
 		public final List<MapRecord<K, HK, HV>> doInRedis(RedisConnection connection) {
 
 			List<ByteRecord> raw = inRedis(connection);
+			if(raw == null) {
+				return Collections.emptyList();
+			}
 
 			List<MapRecord<K, HK, HV>> result = new ArrayList<>();
 			for (ByteRecord record : raw) {
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java
index d33273f46..456e27437 100644
--- a/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/ReactiveStreamOperations.java
@@ -164,7 +164,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
 	/**
 	 * Create a consumer group at the {@link ReadOffset#latest() latest offset}.
 	 *
-	 * @param key
+	 * @param key the {@literal key} the stream is stored at.
 	 * @param group name of the consumer group.
 	 * @return the {@link Mono} emitting {@literal OK} if successful.. {@literal null} when used in pipeline /
 	 *         transaction.
@@ -176,8 +176,8 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
 	/**
 	 * Create a consumer group.
 	 *
-	 * @param key
-	 * @param readOffset
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param readOffset the {@link ReadOffset} to apply.
 	 * @param group name of the consumer group.
 	 * @return the {@link Mono} emitting {@literal OK} if successful.
 	 */
@@ -270,6 +270,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
 	 * @return the {@link Flux} emitting records one by one.
 	 * @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
 	 */
+	@SuppressWarnings("unchecked")
 	default Flux<MapRecord<K, HK, HV>> read(StreamOffset<K> stream) {
 
 		Assert.notNull(stream, "StreamOffset must not be null");
@@ -285,6 +286,7 @@ public interface ReactiveStreamOperations<K, HK, HV> extends HashMapperProvider<
 	 * @return the {@link Flux} emitting records one by one.
 	 * @see <a href="http://redis.io/commands/xread">Redis Documentation: XREAD</a>
 	 */
+	@SuppressWarnings("unchecked")
 	default <V> Flux<ObjectRecord<K, V>> read(Class<V> targetType, StreamOffset<K> stream) {
 
 		Assert.notNull(stream, "StreamOffset must not be null");
diff --git a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java
index aec5b2130..574b1aedb 100644
--- a/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java
+++ b/src/main/java/org/springframework/data/redis/core/StreamObjectMapper.java
@@ -137,7 +137,7 @@ class StreamObjectMapper {
 	 * single-element and multi-element list transformation.l
 	 * 
 	 * @param records the {@link MapRecord} that should be mapped.
-	 * @param hashMapperProvider
+	 * @param hashMapperProvider the provider to obtain the actual {@link HashMapper} from. Must not be {@literal null}.
 	 * @param targetType the requested {@link Class target type}.
 	 * @return the resulting {@link List} of {@link ObjectRecord} or {@literal null} if {@code records} was
 	 *         {@literal null}.
@@ -178,14 +178,17 @@ class StreamObjectMapper {
 	 * 
 	 * @param conversionService the used {@link ConversionService}.
 	 * @param targetType the target type.
-	 * @return
+	 * @return obtain the {@link HashMapper} for a certain type.
 	 */
 	protected HashMapper<?, ?, ?> doGetHashMapper(ConversionService conversionService, Class<?> targetType) {
 		return this.objectHashMapper != null ? objectHashMapper : this.mapper;
 	}
 
 	/**
-	 * @param targetType
+	 * Check if the given type is a simple type as in
+	 * {@link org.springframework.data.convert.CustomConversions#isSimpleType(Class)}.
+	 *
+	 * @param targetType the type to inspect. Must not be {@literal null}.
 	 * @return {@literal true} if {@link Class targetType} is a simple type.
 	 * @see org.springframework.data.convert.CustomConversions#isSimpleType(Class)
 	 */
diff --git a/src/main/java/org/springframework/data/redis/core/StreamOperations.java b/src/main/java/org/springframework/data/redis/core/StreamOperations.java
index 1a81fdc20..16936112b 100644
--- a/src/main/java/org/springframework/data/redis/core/StreamOperations.java
+++ b/src/main/java/org/springframework/data/redis/core/StreamOperations.java
@@ -162,7 +162,7 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
 	/**
 	 * Create a consumer group at the {@link ReadOffset#latest() latest offset}.
 	 *
-	 * @param key
+	 * @param key the {@literal key} the stream is stored at.
 	 * @param group name of the consumer group.
 	 * @return {@literal OK} if successful. {@literal null} when used in pipeline / transaction.
 	 */
@@ -173,8 +173,8 @@ public interface StreamOperations<K, HK, HV> extends HashMapperProvider<HK, HV>
 	/**
 	 * Create a consumer group.
 	 *
-	 * @param key
-	 * @param readOffset
+	 * @param key the {@literal key} the stream is stored at.
+	 * @param readOffset the {@link ReadOffset} to apply.
 	 * @param group name of the consumer group.
 	 * @return {@literal OK} if successful. {@literal null} when used in pipeline / transaction.
 	 */
diff --git a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
index 80be613b2..90d985284 100644
--- a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
+++ b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
@@ -152,33 +152,6 @@ public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
 		return type.cast(fromHash(hash));
 	}
 
-	public Map<String, Object> toObjectHash(Object source) {
-
-		Map<byte[], byte[]> raw = toHash(source);
-
-		RedisPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(source.getClass());
-		Map<String, Object> result = new LinkedHashMap<>();
-
-		for(Map.Entry<byte[], byte[]> entry : raw.entrySet()) {
-
-			String key = converter.fromBytes(entry.getKey(), String.class);
-			Object value = entry.getValue();
-
-			try {
-				value = converter.fromBytes(entry.getValue(), PropertyPath.from(key, entity.getTypeInformation()).getType());
-			} catch (PropertyReferenceException e) {
-				value = converter.fromBytes(entry.getValue(), String.class);
-			} catch (ConverterNotFoundException cnfe) {
-//				value = fromHash(entry)
-				// TODO: nested ones!
-			}
-
-			result.put(key, value);
-		}
-
-		return result;
-	}
-
 	/**
 	 * {@link ReferenceResolver} implementation always returning an empty {@link Map}.
 	 *
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 8de6a2c22..004885d65 100644
--- a/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
+++ b/src/main/java/org/springframework/data/redis/stream/DefaultStreamReceiver.java
@@ -193,8 +193,6 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 
 		/**
 		 * Arm the subscription so {@link Subscription#request(long) demand} activates polling.
-		 *
-		 * @return
 		 */
 		void arm() {
 
@@ -456,8 +454,8 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		/**
 		 * Create a new state object for standalone-read.
 		 *
-		 * @param offset
-		 * @return
+		 * @param offset the {@link ReadOffset} from where to start. Must not be {@literal null}.
+		 * @return new instance of {@link PollState}.
 		 */
 		static PollState standalone(ReadOffset offset) {
 
@@ -468,9 +466,9 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		/**
 		 * Create a new state object for consumergroup-read.
 		 *
-		 * @param consumer
-		 * @param offset
-		 * @return
+		 * @param consumer the {@link Consumer}. Must not be {@literal null}.
+		 * @param offset the {@link ReadOffset} from where to start. Must not be {@literal null}.
+		 * @return new instance of {@link PollState}.
 		 */
 		static PollState consumer(Consumer consumer, ReadOffset offset) {
 
@@ -496,7 +494,7 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		/**
 		 * Decrement request count to indicate that an element was emitted.
 		 *
-		 * @return
+		 * @return {@literal true} if there are still requests pending and the decrement has been successful.
 		 */
 		boolean decrementRequested() {
 
@@ -526,9 +524,9 @@ class DefaultStreamReceiver<K, V extends Record<K, ?>> implements StreamReceiver
 		/**
 		 * Update demand.
 		 *
-		 * @param expect
-		 * @param update
-		 * @return
+		 * @param expect the number of requests still pending.
+		 * @param update the increment to apply
+		 * @return {@literal true} if successful.
 		 */
 		boolean setRequested(long expect, long update) {
 			return requestsPending.compareAndSet(expect, update);
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 de7aa8706..2aea10cda 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamMessageListenerContainer.java
@@ -511,7 +511,7 @@ public interface StreamMessageListenerContainer<K, V extends Record<K, ?>> exten
 		/**
 		 * Timeout for blocking polling using the {@code BLOCK} option during reads.
 		 *
-		 * @return
+		 * @return the timeout.
 		 */
 		public Duration getPollTimeout() {
 			return pollTimeout;
diff --git a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java
index c1cce80a7..f1cd4abde 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamPollTask.java
@@ -184,8 +184,8 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
 		/**
 		 * Create a new state object for standalone-read.
 		 *
-		 * @param offset
-		 * @return
+		 * @param offset the {@link ReadOffset} to use.
+		 * @return new instance of {@link PollState}.
 		 */
 		static PollState standalone(ReadOffset offset) {
 
@@ -196,9 +196,9 @@ class StreamPollTask<K, V extends Record<K, ?>> implements Task {
 		/**
 		 * Create a new state object for consumergroup-read.
 		 *
-		 * @param consumer
-		 * @param offset
-		 * @return
+		 * @param consumer the {@link Consumer} to use.
+		 * @param offset the {@link ReadOffset} to apply.
+		 * @return new instance of {@link PollState}.
 		 */
 		static PollState consumer(Consumer consumer, ReadOffset offset) {
 
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 de196aa08..b2707a3ea 100644
--- a/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
+++ b/src/main/java/org/springframework/data/redis/stream/StreamReceiver.java
@@ -194,6 +194,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		private final @Nullable Class<Object> targetType;
 		private final @Nullable HashMapper<Object, Object, Object> hashMapper;
 
+		@SuppressWarnings({ "unchecked", "rawtypes" })
 		private StreamReceiverOptions(Duration pollTimeout, @Nullable Integer batchSize, SerializationPair<K> keySerializer,
 				SerializationPair<Object> hashKeySerializer, SerializationPair<Object> hashValueSerializer,
 				@Nullable Class<?> targetType, @Nullable HashMapper<V, ?, ?> hashMapper) {
@@ -232,7 +233,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		/**
 		 * Timeout for blocking polling using the {@code BLOCK} option during reads.
 		 *
-		 * @return
+		 * @return the actual timeout.
 		 */
 		public Duration getPollTimeout() {
 			return pollTimeout;
@@ -326,6 +327,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		 * @param pair must not be {@literal null}.
 		 * @return {@code this} {@link StreamReceiverOptionsBuilder}.
 		 */
+		@SuppressWarnings("unchecked")
 		public <T> StreamReceiverOptionsBuilder<T, MapRecord<T, T, T>> serializer(SerializationPair<T> pair) {
 
 			Assert.notNull(pair, "SerializationPair must not be null");
@@ -339,9 +341,10 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		/**
 		 * Configure a key, hash key and hash value serializer.
 		 *
-		 * @param pair must not be {@literal null}.
+		 * @param serializationContext must not be {@literal null}.
 		 * @return {@code this} {@link StreamReceiverOptionsBuilder}.
 		 */
+		@SuppressWarnings("unchecked")
 		public <T> StreamReceiverOptionsBuilder<T, MapRecord<T, T, T>> serializer(
 				RedisSerializationContext<T, ?> serializationContext) {
 
@@ -360,6 +363,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		 * @param pair must not be {@literal null}.
 		 * @return {@code this} {@link StreamReceiverOptionsBuilder}.
 		 */
+		@SuppressWarnings("unchecked")
 		public <NK, NV extends Record<NK, ?>> StreamReceiverOptionsBuilder<NK, NV> keySerializer(
 				SerializationPair<NK> pair) {
 
@@ -375,6 +379,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		 * @param pair must not be {@literal null}.
 		 * @return {@code this} {@link StreamReceiverOptionsBuilder}.
 		 */
+		@SuppressWarnings("unchecked")
 		public <HK, HV> StreamReceiverOptionsBuilder<K, MapRecord<K, HK, HV>> hashKeySerializer(
 				SerializationPair<HK> pair) {
 
@@ -390,6 +395,7 @@ public interface StreamReceiver<K, V extends Record<K, ?>> {
 		 * @param pair must not be {@literal null}.
 		 * @return {@code this} {@link StreamReceiverOptionsBuilder}.
 		 */
+		@SuppressWarnings("unchecked")
 		public <HK, HV> StreamReceiverOptionsBuilder<K, MapRecord<K, HK, HV>> hashValueSerializer(
 				SerializationPair<HK> pair) {