Ensure compatibility with MongoDB Driver 5.x

Jenkinsfile

@@ -265,6 +265,34 @@ pipeline {
 					}
 				}
 
+				stage("test: MongoDB 7.0 (driver-next)") {
+                	agent {
+                		label 'data'
+                	}
+                	options { timeout(time: 30, unit: 'MINUTES') }
+                	environment {
+                		ARTIFACTORY = credentials("${p['artifactory.credentials']}")
+                		DEVELOCITY_CACHE = credentials("${p['develocity.cache.credentials']}")
+                		DEVELOCITY_ACCESS_KEY = credentials("${p['develocity.access-key']}")
+                	}
+                	steps {
+                		script {
+                			docker.image("harbor-repo.vmware.com/dockerhub-proxy-cache/springci/spring-data-with-mongodb-7.0:${p['java.main.tag']}").inside(p['docker.java.inside.basic']) {
+                				sh 'mkdir -p /tmp/mongodb/db /tmp/mongodb/log'
+                				sh 'mongod --setParameter transactionLifetimeLimitSeconds=90 --setParameter maxTransactionLockRequestTimeoutMillis=10000 --dbpath /tmp/mongodb/db --replSet rs0 --fork --logpath /tmp/mongodb/log/mongod.log &'
+                				sh 'sleep 10'
+                				sh 'mongosh --eval "rs.initiate({_id: \'rs0\', members:[{_id: 0, host: \'127.0.0.1:27017\'}]});"'
+                				sh 'sleep 15'
+                				sh 'MAVEN_OPTS="-Duser.name=' + "${p['jenkins.user.name']}" + ' -Duser.home=/tmp/jenkins-home" ' +
+                					"DEVELOCITY_CACHE_USERNAME=${DEVELOCITY_CACHE_USR} " +
+                					"DEVELOCITY_CACHE_PASSWORD=${DEVELOCITY_CACHE_PSW} " +
+                					"GRADLE_ENTERPRISE_ACCESS_KEY=${DEVELOCITY_ACCESS_KEY} " +
+                					"./mvnw -s settings.xml -Pmongo-5.0 clean dependency:list test -Dsort -U -B -Dgradle.cache.local.enabled=false -Dgradle.cache.remote.enabled=false"
+                			}
+                		}
+                	}
+                }
+
 				stage("test: MongoDB 7.0 (next)") {
 					agent {
 						label 'data'

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.3.0-SNAPSHOT</version>
+	<version>4.3.x-4578-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>
@@ -132,6 +132,13 @@
 				<module>spring-data-mongodb-benchmarks</module>
 			</modules>
 		</profile>
+		<profile>
+			<id>mongo-5.0</id>
+			<properties>
+				<mongo>5.0.0-beta0</mongo>
+			</properties>
+		</profile>
+
 	</profiles>
 
 	<dependencies>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.3.0-SNAPSHOT</version>
+		<version>4.3.x-4578-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.3.0-SNAPSHOT</version>
+		<version>4.3.x-4578-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.3.0-SNAPSHOT</version>
+		<version>4.3.x-4578-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -260,6 +260,12 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>org.junit.platform</groupId>
+			<artifactId>junit-platform-launcher</artifactId>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>jakarta.transaction</groupId>
 			<artifactId>jakarta.transaction-api</artifactId>

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/aot/MongoAotPredicates.java

@@ -25,6 +25,9 @@
 import org.springframework.util.ClassUtils;
 
 /**
+ * Collection of {@link Predicate predicates} to determine dynamic library aspects during AOT computation.
+ * Intended for internal usage only.
+ *
  * @author Christoph Strobl
  * @since 4.0
  */
@@ -33,13 +36,27 @@ public class MongoAotPredicates {
 	public static final Predicate<Class<?>> IS_SIMPLE_TYPE = (type) ->  MongoSimpleTypes.HOLDER.isSimpleType(type) || TypeUtils.type(type).isPartOf("org.bson");
 	public static final Predicate<ReactiveLibrary> IS_REACTIVE_LIBARARY_AVAILABLE = ReactiveWrappers::isAvailable;
 	public static final Predicate<ClassLoader> IS_SYNC_CLIENT_PRESENT = (classLoader) -> ClassUtils.isPresent("com.mongodb.client.MongoClient", classLoader);
+	public static final Predicate<ClassLoader> IS_REACTIVE_CLIENT_PRESENT = (classLoader) -> ClassUtils.isPresent("com.mongodb.reactivestreams.client.MongoClient", classLoader);
 
 	public static boolean isReactorPresent() {
 		return IS_REACTIVE_LIBARARY_AVAILABLE.test(ReactiveWrappers.ReactiveLibrary.PROJECT_REACTOR);
 	}
 
+	/**
+	 * @param classLoader can be {@literal null}.
+	 * @return {@literal true} if the {@link com.mongodb.client.MongoClient} is present.
+	 * @since 4.0
+	 */
 	public static boolean isSyncClientPresent(@Nullable ClassLoader classLoader) {
 		return IS_SYNC_CLIENT_PRESENT.test(classLoader);
 	}
 
+	/**
+	 * @param classLoader can be {@literal null}.
+	 * @return {@literal true} if the {@link com.mongodb.reactivestreams.client.MongoClient} is present.
+	 * @since 4.3
+	 */
+	public static boolean isReactiveClientPresent(@Nullable ClassLoader classLoader) {
+		return IS_REACTIVE_CLIENT_PRESENT.test(classLoader);
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/aot/MongoRuntimeHints.java

@@ -19,6 +19,13 @@
 
 import java.util.Arrays;
 
+import com.mongodb.MongoClientSettings;
+import com.mongodb.ServerAddress;
+import com.mongodb.UnixServerAddress;
+import com.mongodb.client.MapReduceIterable;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.IndexOptions;
+import com.mongodb.reactivestreams.client.MapReducePublisher;
 import org.springframework.aot.hint.MemberCategory;
 import org.springframework.aot.hint.RuntimeHints;
 import org.springframework.aot.hint.RuntimeHintsRegistrar;
@@ -31,6 +38,7 @@
 import org.springframework.data.mongodb.core.mapping.event.ReactiveAfterSaveCallback;
 import org.springframework.data.mongodb.core.mapping.event.ReactiveBeforeConvertCallback;
 import org.springframework.data.mongodb.core.mapping.event.ReactiveBeforeSaveCallback;
+import org.springframework.data.mongodb.util.MongoClientVersion;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
 
@@ -53,6 +61,7 @@ public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader)
 						MemberCategory.INVOKE_PUBLIC_METHODS));
 
 		registerTransactionProxyHints(hints, classLoader);
+		registerMongoCompatibilityAdapterHints(hints, classLoader);
 
 		if (isReactorPresent()) {
 
@@ -80,4 +89,33 @@ private static void registerTransactionProxyHints(RuntimeHints hints, @Nullable
 		}
 	}
 
+	private static void registerMongoCompatibilityAdapterHints(RuntimeHints hints, @Nullable ClassLoader classLoader) {
+
+		hints.reflection() //
+				.registerType(MongoClientSettings.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+				.registerType(MongoClientSettings.Builder.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+				.registerType(IndexOptions.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+				.registerType(ServerAddress.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+				.registerType(UnixServerAddress.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+				.registerType(TypeReference.of("com.mongodb.connection.StreamFactoryFactory"), MemberCategory.INTROSPECT_PUBLIC_METHODS);
+
+		if(MongoAotPredicates.isSyncClientPresent(classLoader)) {
+
+			hints.reflection() //
+					.registerType(MongoDatabase.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(TypeReference.of("com.mongodb.client.internal.MongoDatabaseImpl"), MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(MapReduceIterable.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(TypeReference.of("com.mongodb.client.internal.MapReduceIterableImpl"), MemberCategory.INVOKE_PUBLIC_METHODS);
+		}
+
+		if(MongoAotPredicates.isReactiveClientPresent(classLoader)) {
+
+			hints.reflection() //
+					.registerType(com.mongodb.reactivestreams.client.MongoDatabase.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(TypeReference.of("com.mongodb.reactivestreams.client.internal.MongoDatabaseImpl"), MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(MapReducePublisher.class, MemberCategory.INVOKE_PUBLIC_METHODS)
+					.registerType(TypeReference.of("com.mongodb.reactivestreams.client.internal.MapReducePublisherImpl"), MemberCategory.INVOKE_PUBLIC_METHODS);
+		}
+	}
+
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultIndexOperations.java

@@ -114,6 +114,7 @@ public DefaultIndexOperations(MongoOperations mongoOperations, String collection
 		this.type = type;
 	}
 
+	@Override
 	public String ensureIndex(final IndexDefinition indexDefinition) {
 
 		return execute(collection -> {

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/DefaultReactiveIndexOperations.java

@@ -88,7 +88,8 @@ private DefaultReactiveIndexOperations(ReactiveMongoOperations mongoOperations,
 		this.type = type;
 	}
 
-	public Mono<String> ensureIndex(final IndexDefinition indexDefinition) {
+	@Override
+	public Mono<String> ensureIndex(IndexDefinition indexDefinition) {
 
 		return mongoOperations.execute(collectionName, collection -> {
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/IndexConverters.java

@@ -18,9 +18,11 @@
 import java.util.concurrent.TimeUnit;
 
 import org.bson.Document;
+
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.mongodb.core.index.IndexDefinition;
 import org.springframework.data.mongodb.core.index.IndexInfo;
+import org.springframework.data.mongodb.util.MongoCompatibilityAdapter;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 
@@ -89,7 +91,7 @@ private static Converter<IndexDefinition, IndexOptions> getIndexDefinitionIndexO
 				ops = ops.bits((Integer) indexOptions.get("bits"));
 			}
 			if (indexOptions.containsKey("bucketSize")) {
-				ops = ops.bucketSize(((Number) indexOptions.get("bucketSize")).doubleValue());
+				MongoCompatibilityAdapter.indexOptionsAdapter(ops).setBucketSize(((Number) indexOptions.get("bucketSize")).doubleValue());
 			}
 			if (indexOptions.containsKey("default_language")) {
 				ops = ops.defaultLanguage(indexOptions.get("default_language").toString());

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoClientSettingsFactoryBean.java

@@ -25,7 +25,9 @@
 
 import org.bson.UuidRepresentation;
 import org.bson.codecs.configuration.CodecRegistry;
+
 import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.data.mongodb.util.MongoCompatibilityAdapter;
 import org.springframework.lang.Nullable;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
@@ -40,7 +42,7 @@
 import com.mongodb.WriteConcern;
 import com.mongodb.connection.ClusterConnectionMode;
 import com.mongodb.connection.ClusterType;
-import com.mongodb.connection.StreamFactoryFactory;
+import com.mongodb.connection.TransportSettings;
 
 /**
  * A factory bean for construction of a {@link MongoClientSettings} instance to be used with a MongoDB driver.
@@ -54,7 +56,10 @@ public class MongoClientSettingsFactoryBean extends AbstractFactoryBean<MongoCli
 	private static final MongoClientSettings DEFAULT_MONGO_SETTINGS = MongoClientSettings.builder().build();
 
 	private CodecRegistry codecRegistry = DEFAULT_MONGO_SETTINGS.getCodecRegistry();
-	private StreamFactoryFactory streamFactoryFactory = DEFAULT_MONGO_SETTINGS.getStreamFactoryFactory();
+
+	@Nullable private Object streamFactoryFactory = MongoCompatibilityAdapter
+			.clientSettingsAdapter(DEFAULT_MONGO_SETTINGS).getStreamFactoryFactory();
+	@Nullable private TransportSettings transportSettings;
 
 	private ReadPreference readPreference = DEFAULT_MONGO_SETTINGS.getReadPreference();
 	private ReadConcern readConcern = DEFAULT_MONGO_SETTINGS.getReadConcern();
@@ -118,15 +123,15 @@ public class MongoClientSettingsFactoryBean extends AbstractFactoryBean<MongoCli
 
 	/**
 	 * @param socketConnectTimeoutMS in msec
-	 * @see com.mongodb.connection.SocketSettings.Builder#connectTimeout(int, TimeUnit)
+	 * @see com.mongodb.connection.SocketSettings.Builder
 	 */
 	public void setSocketConnectTimeoutMS(int socketConnectTimeoutMS) {
 		this.socketConnectTimeoutMS = socketConnectTimeoutMS;
 	}
 
 	/**
 	 * @param socketReadTimeoutMS in msec
-	 * @see com.mongodb.connection.SocketSettings.Builder#readTimeout(int, TimeUnit)
+	 * @see com.mongodb.connection.SocketSettings.Builder
 	 */
 	public void setSocketReadTimeoutMS(int socketReadTimeoutMS) {
 		this.socketReadTimeoutMS = socketReadTimeoutMS;
@@ -368,12 +373,18 @@ public void setReadPreference(ReadPreference readPreference) {
 
 	/**
 	 * @param streamFactoryFactory
-	 * @see MongoClientSettings.Builder#streamFactoryFactory(StreamFactoryFactory)
+	 * @deprecated since 4.3, will be removed in the MongoDB 5.0 driver in favor of
+	 *             {@code com.mongodb.connection.TransportSettings}.
 	 */
-	public void setStreamFactoryFactory(StreamFactoryFactory streamFactoryFactory) {
+	@Deprecated(since = "4.3")
+	public void setStreamFactoryFactory(Object streamFactoryFactory) {
 		this.streamFactoryFactory = streamFactoryFactory;
 	}
 
+	public void setTransportSettings(@Nullable TransportSettings transportSettings) {
+		this.transportSettings = transportSettings;
+	}
+
 	/**
 	 * @param codecRegistry
 	 * @see MongoClientSettings.Builder#codecRegistry(CodecRegistry)
@@ -433,7 +444,6 @@ protected MongoClientSettings createInstance() {
 						settings.hosts(clusterHosts);
 					}
 					settings.localThreshold(clusterLocalThresholdMS, TimeUnit.MILLISECONDS);
-					// settings.maxWaitQueueSize(clusterMaxWaitQueueSize);
 					settings.requiredClusterType(custerRequiredClusterType);
 
 					if (StringUtils.hasText(clusterSrvHost)) {
@@ -478,12 +488,18 @@ protected MongoClientSettings createInstance() {
 					}
 				});
 
+		if (transportSettings != null) {
+			builder.transportSettings(transportSettings);
+		}
+
 		if (streamFactoryFactory != null) {
-			builder = builder.streamFactoryFactory(streamFactoryFactory);
+			MongoCompatibilityAdapter.clientSettingsBuilderAdapter(builder).setStreamFactoryFactory(streamFactoryFactory);
 		}
+
 		if (retryReads != null) {
 			builder = builder.retryReads(retryReads);
 		}
+
 		if (retryWrites != null) {
 			builder = builder.retryWrites(retryWrites);
 		}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -30,6 +30,7 @@
 import org.apache.commons.logging.LogFactory;
 import org.bson.Document;
 import org.bson.conversions.Bson;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -103,6 +104,7 @@
 import org.springframework.data.mongodb.core.query.UpdateDefinition.ArrayFilter;
 import org.springframework.data.mongodb.core.timeseries.Granularity;
 import org.springframework.data.mongodb.core.validation.Validator;
+import org.springframework.data.mongodb.util.MongoCompatibilityAdapter;
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.data.util.Optionals;
@@ -722,7 +724,7 @@ public boolean collectionExists(String collectionName) {
 
 		return execute(db -> {
 
-			for (String name : db.listCollectionNames()) {
+			for (String name : MongoCompatibilityAdapter.mongoDatabaseAdapter().forDb(db).listCollectionNames()) {
 				if (name.equals(collectionName)) {
 					return true;
 				}
@@ -1965,7 +1967,7 @@ public <T> List<T> mapReduce(Query query, Class<?> domainType, String inputColle
 			}
 
 			if (mapReduceOptions.getOutputSharded().isPresent()) {
-				mapReduce = mapReduce.sharded(mapReduceOptions.getOutputSharded().get());
+				MongoCompatibilityAdapter.mapReduceIterableAdapter(mapReduce).sharded(mapReduceOptions.getOutputSharded().get());
 			}
 
 			if (StringUtils.hasText(mapReduceOptions.getOutputCollection()) && !mapReduceOptions.usesInlineOutput()) {
@@ -2340,7 +2342,7 @@ protected String replaceWithResourceIfNecessary(String function) {
 	public Set<String> getCollectionNames() {
 		return execute(db -> {
 			Set<String> result = new LinkedHashSet<>();
-			for (String name : db.listCollectionNames()) {
+			for (String name : MongoCompatibilityAdapter.mongoDatabaseAdapter().forDb(db).listCollectionNames()) {
 				result.add(name);
 			}
 			return result;