diff --git a/src/main/java/org/springframework/data/repository/query/ResultProcessor.java b/src/main/java/org/springframework/data/repository/query/ResultProcessor.java
index c37452168..e65eaf3ce 100644
--- a/src/main/java/org/springframework/data/repository/query/ResultProcessor.java
+++ b/src/main/java/org/springframework/data/repository/query/ResultProcessor.java
@@ -40,6 +40,7 @@ import org.springframework.util.Assert;
  * query results into projections and data transfer objects.
  * 
  * @author Oliver Gierke
+ * @author John Blum
  * @since 1.12
  */
 public class ResultProcessor {
@@ -152,14 +153,7 @@ public class ResultProcessor {
 		}
 
 		if (ReflectionUtils.isJava8StreamType(source.getClass()) && method.isStreamQuery()) {
-
-			return (T) ((Stream<Object>) source).map(new Function<Object, T>() {
-
-				@Override
-				public T apply(Object t) {
-					return (T) (type.isInstance(t) ? t : converter.convert(t));
-				}
-			});
+			return (T) new StreamQueryResultHandler(type, converter).handle(source);
 		}
 
 		return (T) converter.convert(source);
@@ -271,4 +265,39 @@ public class ResultProcessor {
 			return result;
 		}
 	}
+
+	/**
+	 * Handler for Repository query methods returning a Java 8 Stream result by ensuring the {@link Stream} elements match
+	 * the expected return type of the query method.
+	 *
+	 * @author John Blum
+	 * @author Oliver Gierke
+	 */
+	@RequiredArgsConstructor
+	static class StreamQueryResultHandler {
+
+		private final @NonNull ReturnedType returnType;
+		private final @NonNull Converter<Object, Object> converter;
+
+		/**
+		 * Processes the given source object as a {@link Stream}, mapping each element to the required return type,
+		 * converting if necessary.
+		 *
+		 * @param source the {@link Stream} of elements to process, must not be {@literal null}.
+		 * @return a new {@link Stream} with the source {@link Stream}'s elements mapped to the target type.
+		 */
+		@SuppressWarnings("unchecked")
+		public Object handle(Object source) {
+
+			Assert.isInstanceOf(Stream.class, source, "Source must not be null and an instance of Stream!");
+
+			return ((Stream<Object>) source).map(new Function<Object, Object>() {
+
+				@Override
+				public Object apply(Object element) {
+					return returnType.isInstance(element) ? element : converter.convert(element);
+				}
+			});
+		}
+	}
 }
diff --git a/src/test/java/org/springframework/data/repository/query/StreamQueryResultHandlerUnitTests.java b/src/test/java/org/springframework/data/repository/query/StreamQueryResultHandlerUnitTests.java
new file mode 100644
index 000000000..54f9850a4
--- /dev/null
+++ b/src/test/java/org/springframework/data/repository/query/StreamQueryResultHandlerUnitTests.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2016 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
+ *
+ *      http://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.repository.query;
+
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import lombok.Value;
+
+import java.util.Arrays;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.repository.query.ResultProcessor.StreamQueryResultHandler;
+
+/**
+ * Unit tests for {@link StreamQueryResultHandler}.
+ *
+ * @author John Blum
+ * @author Oliver Gierke
+ */
+public class StreamQueryResultHandlerUnitTests {
+
+	StreamQueryResultHandler handler;
+
+	@Before
+	public void setUp() {
+
+		ReturnedType returnedType = ReturnedType.of(String.class, Person.class, mock(ProjectionFactory.class));
+
+		this.handler = new StreamQueryResultHandler(returnedType, new Converter<Object, Object>() {
+
+			@Override
+			public Object convert(Object source) {
+				return source.toString();
+			}
+		});
+	}
+
+	/**
+	 * @see DATACMNS-868
+	 */
+	@Test
+	@SuppressWarnings("unchecked")
+	public void mapsStreamUsingConverter() {
+
+		Stream<Person> people = Arrays.asList(Person.of("Dave", "Matthews")).stream();
+
+		Object result = this.handler.handle(people);
+
+		assertThat(result, is(instanceOf(Stream.class)));
+
+		Stream<Object> stream = (Stream<Object>) result;
+
+		assertThat(stream.allMatch(new Predicate<Object>() {
+
+			@Override
+			public boolean test(Object t) {
+
+				assertThat(t, is(instanceOf(String.class)));
+
+				String string = (String) t;
+
+				assertThat(string, containsString("Dave"));
+				assertThat(string, containsString("Matthews"));
+
+				return true;
+			}
+		}), is(true));
+
+		stream.close();
+	}
+
+	/**
+	 * @see DATACMNS-868
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void rejectsNullSource() {
+		handler.handle(null);
+	}
+
+	@Value(staticConstructor = "of")
+	static class Person {
+		String firstName, lastName;
+	}
+}