GH-2603 - Upgrade to Neo4j Java Driver 5.x.

This is the first of two changes in SDN 7 regarding Neo4j 5: It brings in the new driver 5.0.0 and adapts to its new API in two steps:

- All calls to deprecateed API are replaced to the new API where possible.
- All calls to `id()` are isolated in a separate utility class
- The internal ID generator now logs a warning when used

A follow up change will introduce a new internal id generator supporting the new element ids.

Nothing is changed internally, SDN still relies on the long id values.

Closes #2603

Author: michael-simons

pom.xml

@@ -76,7 +76,7 @@
 		<cdi>3.0.1</cdi>
 		<checkstyle.skip>${skipTests}</checkstyle.skip>
 		<checkstyle.version>8.40</checkstyle.version>
-		<cypher-dsl.version>2022.4.0</cypher-dsl.version>
+		<cypher-dsl.version>2022.6.0</cypher-dsl.version>
 		<dist.id>spring-data-neo4j</dist.id>
 		<dist.key>SDNEO4J</dist.key>
 		<flatten-maven-plugin.version>1.2.5</flatten-maven-plugin.version>
@@ -104,8 +104,9 @@
 		<maven.compiler.release>${java.version}</maven.compiler.release>
 		<mockito>${mockito.version}</mockito>
 		<mockito.version>4.4.0</mockito.version>
-		<neo4j-java-driver.version>4.4.9</neo4j-java-driver.version>
-		<neo4j.version>4.3.6</neo4j.version>
+		<neo4j-java-driver.version>5.0.0</neo4j-java-driver.version>
+		<neo4j-migrations.version>1.8.1</neo4j-migrations.version>
+		<neo4j.version>4.4.8</neo4j.version>
 		<objenesis.version>3.0.1</objenesis.version>
 		<project.build.docs>${project.build.directory}/docs</project.build.docs>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -238,6 +239,16 @@
 			<artifactId>jackson-databind</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<!-- Including this explicitly in test scope makes
+			[WARNING] unknown enum constant When.MAYBE
+			[WARNING]   reason: class file for javax.annotation.meta.When not found
+			go away
+			-->
+			<groupId>com.google.code.findbugs</groupId>
+			<artifactId>jsr305</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>com.querydsl</groupId>
 			<artifactId>querydsl-core</artifactId>
@@ -255,6 +266,12 @@
 			<artifactId>junit-jupiter-causal-cluster-testcontainer-extension</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>eu.michael-simons.neo4j</groupId>
+			<artifactId>neo4j-migrations</artifactId>
+			<version>${neo4j-migrations.version}</version>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.mockk</groupId>
 			<artifactId>mockk-jvm</artifactId>
@@ -739,16 +756,25 @@
 				<artifactId>maven-compiler-plugin</artifactId>
 				<executions>
 					<execution>
-						<goals>
-							<goal>compile</goal>
-						</goals>
+						<id>java-compile</id>
 						<configuration combine.self="append">
 							<compilerArgs>
 								<arg>-Xlint:all,-options,-path</arg>
 								<arg>-Werror</arg>
 							</compilerArgs>
 						</configuration>
 					</execution>
+					<execution>
+						<!-- You would think that applying above configuration to java-test-compile would have the same result.
+						     Alas, it doesn't. The build just fails, -e -X doesn't show anything sane.
+						     So, showDeprecation is just warnings, if set to errors, than it fails hard (which is nice for
+						     deprecation, but it also fails during missing When.MAYBE whateever Jakarta annotation
+						     and there's just no motiviation to deal with Spring Data Commons build setup in this project -->
+						<id>java-test-compile</id>
+						<configuration combine.self="append">
+							<showDeprecation>true</showDeprecation>
+						</configuration>
+					</execution>
 				</executions>
 			</plugin>
 		</plugins>

src/main/java/org/springframework/data/neo4j/core/DefaultNeo4jClient.java

@@ -99,13 +99,13 @@ public QueryRunner getQueryRunner(DatabaseSelection databaseSelection, UserSelec
 			}
 		}
 
-		return new DelegatingQueryRunner(queryRunner, lastBookmarks, (usedBookmarks, newBookmark) -> {
+		return new DelegatingQueryRunner(queryRunner, lastBookmarks, (usedBookmarks, newBookmarks) -> {
 
 			ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
 			try {
 				lock.lock();
 				bookmarks.removeAll(usedBookmarks);
-				bookmarks.add(newBookmark);
+				bookmarks.addAll(newBookmarks);
 			} finally {
 				lock.unlock();
 			}
@@ -116,24 +116,23 @@ private static class DelegatingQueryRunner implements QueryRunner {
 
 		private final QueryRunner delegate;
 		private final Collection<Bookmark> usedBookmarks;
-		private final BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer;
+		private final BiConsumer<Collection<Bookmark>, Collection<Bookmark>> newBookmarkConsumer;
 
-		private DelegatingQueryRunner(QueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer) {
+		private DelegatingQueryRunner(QueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Collection<Bookmark>> newBookmarkConsumer) {
 			this.delegate = delegate;
 			this.usedBookmarks = lastBookmarks;
 			this.newBookmarkConsumer = newBookmarkConsumer;
 		}
 
 		@Override
-		public void close() throws Exception {
+		public void close() {
 
 			// We're only going to close sessions we have acquired inside the client, not something that
 			// has been retrieved from the tx manager.
-			if (this.delegate instanceof Session) {
+			if (this.delegate instanceof Session session) {
 
-				Session session = (Session) this.delegate;
 				session.close();
-				this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmark());
+				this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmarks());
 			}
 		}
 

src/main/java/org/springframework/data/neo4j/core/DefaultReactiveNeo4jClient.java

@@ -20,9 +20,9 @@
 import org.neo4j.driver.Query;
 import org.neo4j.driver.Record;
 import org.neo4j.driver.Value;
-import org.neo4j.driver.reactive.RxQueryRunner;
-import org.neo4j.driver.reactive.RxResult;
-import org.neo4j.driver.reactive.RxSession;
+import org.neo4j.driver.reactive.ReactiveQueryRunner;
+import org.neo4j.driver.reactive.ReactiveResult;
+import org.neo4j.driver.reactive.ReactiveSession;
 import org.neo4j.driver.summary.ResultSummary;
 import org.neo4j.driver.types.TypeSystem;
 import org.springframework.core.convert.ConversionService;
@@ -36,6 +36,7 @@
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import reactor.adapter.JdkFlowAdapter;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -47,6 +48,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Flow.Publisher;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
@@ -86,42 +88,42 @@ final class DefaultReactiveNeo4jClient implements ReactiveNeo4jClient {
 	}
 
 	@Override
-	public Mono<RxQueryRunner> getQueryRunner(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection) {
+	public Mono<ReactiveQueryRunner> getQueryRunner(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection) {
 
 		return databaseSelection.zipWith(userSelection)
 				.flatMap(targetDatabaseAndUser ->
-				ReactiveNeo4jTransactionManager.retrieveReactiveTransaction(driver, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())
-						.map(RxQueryRunner.class::cast)
-						.zipWith(Mono.just(Collections.<Bookmark>emptySet()))
-						.switchIfEmpty(Mono.fromSupplier(() -> {
-							ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
-							try {
-								lock.lock();
-								Set<Bookmark> lastBookmarks = new HashSet<>(bookmarks);
-								return Tuples.of(driver.rxSession(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())), lastBookmarks);
-							} finally {
-								lock.unlock();
-							}
-						})))
-				.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), (usedBookmarks, newBookmark) -> {
+						ReactiveNeo4jTransactionManager.retrieveReactiveTransaction(driver, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())
+								.map(ReactiveQueryRunner.class::cast)
+								.zipWith(Mono.just(Collections.<Bookmark>emptySet()))
+								.switchIfEmpty(Mono.fromSupplier(() -> {
+									ReentrantReadWriteLock.ReadLock lock = bookmarksLock.readLock();
+									try {
+										lock.lock();
+										Set<Bookmark> lastBookmarks = new HashSet<>(bookmarks);
+										return Tuples.of(driver.reactiveSession(Neo4jTransactionUtils.sessionConfig(false, lastBookmarks, targetDatabaseAndUser.getT1(), targetDatabaseAndUser.getT2())), lastBookmarks);
+									} finally {
+										lock.unlock();
+									}
+								})))
+				.map(t -> new DelegatingQueryRunner(t.getT1(), t.getT2(), (usedBookmarks, newBookmarks) -> {
 					ReentrantReadWriteLock.WriteLock lock = bookmarksLock.writeLock();
 					try {
 						lock.lock();
 						bookmarks.removeAll(usedBookmarks);
-						bookmarks.add(newBookmark);
+						bookmarks.addAll(newBookmarks);
 					} finally {
 						lock.unlock();
 					}
 				}));
 	}
 
-	private static class DelegatingQueryRunner implements RxQueryRunner {
+	private static class DelegatingQueryRunner implements ReactiveQueryRunner {
 
-		private final RxQueryRunner delegate;
+		private final ReactiveQueryRunner delegate;
 		private final Collection<Bookmark> usedBookmarks;
-		private final BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer;
+		private final BiConsumer<Collection<Bookmark>, Collection<Bookmark>> newBookmarkConsumer;
 
-		private DelegatingQueryRunner(RxQueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Bookmark> newBookmarkConsumer) {
+		private DelegatingQueryRunner(ReactiveQueryRunner delegate, Collection<Bookmark> lastBookmarks, BiConsumer<Collection<Bookmark>, Collection<Bookmark>> newBookmarkConsumer) {
 			this.delegate = delegate;
 			this.usedBookmarks = lastBookmarks;
 			this.newBookmarkConsumer = newBookmarkConsumer;
@@ -131,47 +133,46 @@ Mono<Void> close() {
 
 			// We're only going to close sessions we have acquired inside the client, not something that
 			// has been retrieved from the tx manager.
-			if (this.delegate instanceof RxSession) {
-				RxSession session = (RxSession) this.delegate;
-				return Mono.fromDirect(session.close()).then().doOnSuccess(signal ->
-						this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmark()));
+			if (this.delegate instanceof ReactiveSession session) {
+				return JdkFlowAdapter.flowPublisherToFlux(session.close()).then().doOnSuccess(signal ->
+						this.newBookmarkConsumer.accept(usedBookmarks, session.lastBookmarks()));
 			}
 
 			return Mono.empty();
 		}
 
 		@Override
-		public RxResult run(String query, Value parameters) {
+		public Publisher<ReactiveResult> run(String query, Value parameters) {
 			return delegate.run(query, parameters);
 		}
 
 		@Override
-		public RxResult run(String query, Map<String, Object> parameters) {
+		public Publisher<ReactiveResult> run(String query, Map<String, Object> parameters) {
 			return delegate.run(query, parameters);
 		}
 
 		@Override
-		public RxResult run(String query, Record parameters) {
+		public Publisher<ReactiveResult> run(String query, Record parameters) {
 			return delegate.run(query, parameters);
 		}
 
 		@Override
-		public RxResult run(String query) {
+		public Publisher<ReactiveResult> run(String query) {
 			return delegate.run(query);
 		}
 
 		@Override
-		public RxResult run(Query query) {
+		public Publisher<ReactiveResult> run(Query query) {
 			return delegate.run(query);
 		}
 	}
 
-	<T> Mono<T> doInQueryRunnerForMono(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection, Function<RxQueryRunner, Mono<T>> func) {
+	<T> Mono<T> doInQueryRunnerForMono(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection, Function<ReactiveQueryRunner, Mono<T>> func) {
 
 		return Mono.usingWhen(getQueryRunner(databaseSelection, userSelection), func, runner -> ((DelegatingQueryRunner) runner).close());
 	}
 
-	<T> Flux<T> doInStatementRunnerForFlux(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection, Function<RxQueryRunner, Flux<T>> func) {
+	<T> Flux<T> doInStatementRunnerForFlux(Mono<DatabaseSelection> databaseSelection, Mono<UserSelection> userSelection, Function<ReactiveQueryRunner, Flux<T>> func) {
 
 		return Flux.usingWhen(getQueryRunner(databaseSelection, userSelection), func, runner -> ((DelegatingQueryRunner) runner).close());
 	}
@@ -187,7 +188,7 @@ public UnboundRunnableSpec query(Supplier<String> cypherSupplier) {
 	}
 
 	@Override
-	public <T> OngoingDelegation<T> delegateTo(Function<RxQueryRunner, Mono<T>> callback) {
+	public <T> OngoingDelegation<T> delegateTo(Function<ReactiveQueryRunner, Mono<T>> callback) {
 		return new DefaultRunnableDelegation<>(callback);
 	}
 
@@ -415,27 +416,27 @@ Mono<Tuple2<String, Map<String, Object>>> prepareStatement() {
 			return Mono.fromSupplier(cypherSupplier).zipWith(Mono.just(parameters.get()));
 		}
 
-		Flux<T> executeWith(Tuple2<String, Map<String, Object>> t, RxQueryRunner runner) {
+		Flux<T> executeWith(Tuple2<String, Map<String, Object>> t, ReactiveQueryRunner runner) {
 
-			return Flux.usingWhen(Flux.just(runner.run(t.getT1(), t.getT2())),
-					result -> Flux.from(result.records()).mapNotNull(r -> mappingFunction.apply(typeSystem, r)),
-					result -> Flux.from(result.consume()).doOnNext(ResultSummaries::process));
+			return Flux.usingWhen(JdkFlowAdapter.flowPublisherToFlux(runner.run(t.getT1(), t.getT2())),
+					result -> JdkFlowAdapter.flowPublisherToFlux(result.records()).mapNotNull(r -> mappingFunction.apply(typeSystem, r)),
+					result -> JdkFlowAdapter.flowPublisherToFlux(result.consume()).doOnNext(ResultSummaries::process));
 		}
 
 		@Override
 		public Mono<T> one() {
 
 			return doInQueryRunnerForMono(databaseSelection, userSelection,
-					(runner) -> prepareStatement().flatMapMany(t -> executeWith(t, runner)).singleOrEmpty())
-							.onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException);
+					(runner) -> prepareStatement().flatMapMany(t -> executeWith(t, runner)).singleOrEmpty()
+					.onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException));
 		}
 
 		@Override
 		public Mono<T> first() {
 
 			return doInQueryRunnerForMono(databaseSelection, userSelection,
 					runner -> prepareStatement().flatMapMany(t -> executeWith(t, runner)).next())
-							.onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException);
+					.onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException);
 		}
 
 		@Override
@@ -448,10 +449,10 @@ public Flux<T> all() {
 
 		Mono<ResultSummary> run() {
 
-			return doInQueryRunnerForMono(databaseSelection, userSelection, runner -> prepareStatement().flatMap(t -> {
-				RxResult rxResult = runner.run(t.getT1(), t.getT2());
-				return Flux.from(rxResult.records()).then(Mono.from(rxResult.consume()).map(ResultSummaries::process));
-			})).onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException);
+			return doInQueryRunnerForMono(databaseSelection, userSelection, runner -> prepareStatement()
+					.flatMap(t -> JdkFlowAdapter.flowPublisherToFlux(runner.run(t.getT1(), t.getT2())).single())
+					.flatMap(rxResult -> JdkFlowAdapter.flowPublisherToFlux(rxResult.consume()).single().map(ResultSummaries::process)))
+					.onErrorMap(RuntimeException.class, DefaultReactiveNeo4jClient.this::potentiallyConvertRuntimeException);
 		}
 	}
 
@@ -469,12 +470,12 @@ private RuntimeException potentiallyConvertRuntimeException(RuntimeException ex)
 
 	class DefaultRunnableDelegation<T> implements RunnableDelegation<T>, OngoingDelegation<T> {
 
-		private final Function<RxQueryRunner, Mono<T>> callback;
+		private final Function<ReactiveQueryRunner, Mono<T>> callback;
 
 		private Mono<DatabaseSelection> databaseSelection;
-		private Mono<UserSelection> userSelection;
+		private final Mono<UserSelection> userSelection;
 
-		DefaultRunnableDelegation(Function<RxQueryRunner, Mono<T>> callback) {
+		DefaultRunnableDelegation(Function<ReactiveQueryRunner, Mono<T>> callback) {
 			this.callback = callback;
 			this.databaseSelection = resolveTargetDatabaseName(null);
 			this.userSelection = resolveUser(null);