Fix synchronisation of bookmark access in reactive bookmark manager.

The separate reactive bookmark manager was introduced because Project Reactor and Blockhound have issues with the `ReentrantReadWriteLock` approach we have taken in the imperative approach.

The first approach was using a synchronized set, but as it was correctly noted by @seabamirum on #2769 this is not enough on the reading path:

> It is imperative that the user manually synchronize on the returned set when iterating over it:

(From the JavaDoc of `Collections.synchronizedSet`.

Using `ConcurrentHashMap.newKeySet` would solve that issue, but it would require a check for `null` values in the `usedBookmarks` argument for `updateBookmarks` AND it would also not solve the fact that removing the used bookmarks and adding the new ones is an atomic operation (such as it originally was in the imperative world).

So therefor it is just easier to use a standard set and synchronize over it.

Author: michael-simons

src/main/java/org/springframework/data/neo4j/core/transaction/ReactiveDefaultBookmarkManager.java

@@ -36,7 +36,7 @@
  */
 final class ReactiveDefaultBookmarkManager extends AbstractBookmarkManager {
 
-	private final Set<Bookmark> bookmarks = Collections.synchronizedSet(new HashSet<>());
+	private final Set<Bookmark> bookmarks = new HashSet<>();
 
 	private final Supplier<Set<Bookmark>> bookmarksSupplier;
 
@@ -49,14 +49,18 @@ final class ReactiveDefaultBookmarkManager extends AbstractBookmarkManager {
 
 	@Override
 	public Collection<Bookmark> getBookmarks() {
-		this.bookmarks.addAll(bookmarksSupplier.get());
-		return Set.copyOf(this.bookmarks);
+		synchronized (this.bookmarks) {
+			this.bookmarks.addAll(bookmarksSupplier.get());
+			return Set.copyOf(this.bookmarks);
+		}
 	}
 
 	@Override
 	public void updateBookmarks(Collection<Bookmark> usedBookmarks, Collection<Bookmark> newBookmarks) {
-		bookmarks.removeAll(usedBookmarks);
-		newBookmarks.stream().filter(Objects::nonNull).forEach(bookmarks::add);
+		synchronized (this.bookmarks) {
+			usedBookmarks.stream().filter(Objects::nonNull).forEach(bookmarks::remove);
+			newBookmarks.stream().filter(Objects::nonNull).forEach(bookmarks::add);
+		}
 		if (applicationEventPublisher != null) {
 			applicationEventPublisher.publishEvent(new Neo4jBookmarksUpdatedEvent(new HashSet<>(bookmarks)));
 		}