DATAJPA-1023 - StreamExceution now rejects non-transactional execution.

StreamExecution now makes use of the newly introduced SurroundingTransactionDetectingMethodInterceptor to find out about whether the transaction is used in code that already has a transaction running. This is necessary to make sure the Stream returned by the method can actually be consumed as the surrounding transaction keeps the transaction open.

Related tickets: DATACMNS-959.

Author: odrotbohm

src/main/java/org/springframework/data/jpa/repository/query/JpaQueryExecution.java

@@ -27,10 +27,12 @@
 import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.ConfigurableConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
 import org.springframework.data.jpa.provider.PersistenceProvider;
+import org.springframework.data.repository.core.support.SurroundingTransactionDetectorMethodInterceptor;
 import org.springframework.data.repository.query.ParameterAccessor;
 import org.springframework.data.repository.query.Parameters;
 import org.springframework.data.repository.query.ParametersParameterAccessor;
@@ -323,13 +325,19 @@ protected Object doExecute(AbstractJpaQuery jpaQuery, Object[] values) {
 	 */
 	static class StreamExecution extends JpaQueryExecution {
 
+		private static final String NO_SURROUNDING_TRANSACTION = "You're trying to execute a streaming query method without a surrounding transaction that keeps the connection open so that the Stream can actually be consumed. Make sure the code consuming the stream uses @Transactional or any other way of declaring a (read-only) transaction.";
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.jpa.repository.query.JpaQueryExecution#doExecute(org.springframework.data.jpa.repository.query.AbstractJpaQuery, java.lang.Object[])
 		 */
 		@Override
 		protected Object doExecute(final AbstractJpaQuery query, Object[] values) {
 
+			if (!SurroundingTransactionDetectorMethodInterceptor.INSTANCE.isSurroundingTransactionActive()) {
+				throw new InvalidDataAccessApiUsageException(NO_SURROUNDING_TRANSACTION);
+			}
+
 			Query jpaQuery = query.createQuery(values);
 			PersistenceProvider persistenceProvider = PersistenceProvider.fromEntityManager(query.getEntityManager());
 			CloseableIterator<Object> iter = persistenceProvider.executeQueryWithResultStream(jpaQuery);

src/test/java/org/springframework/data/jpa/repository/UserRepositoryFinderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2015 the original author or authors.
+ * Copyright 2008-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.
@@ -26,6 +26,7 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Slice;
@@ -37,6 +38,7 @@
 import org.springframework.data.repository.query.QueryLookupStrategy;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 
 /**
@@ -234,4 +236,19 @@ public void translatesNotContainsToNotMemberOf() {
 	public void executesQueryWithProjectionContainingReferenceToPluralAttribute() {
 		assertThat(userRepository.findRolesAndFirstnameBy(), is(notNullValue()));
 	}
+
+	/**
+	 * @see DATAJPA-1023, DATACMNS-959
+	 */
+	@Test(expected = InvalidDataAccessApiUsageException.class)
+	@Transactional(propagation = Propagation.NOT_SUPPORTED)
+	public void rejectsStreamExecutionIfNoSurroundingTransactionActive() {
+
+		try {
+			userRepository.findAllByCustomQueryAndStream();
+		} finally {
+			roleRepository.deleteAll();
+			userRepository.deleteAll();
+		}
+	}
 }