feat(mongodb): upgrade MongoDB Java driver to version 5.5.0

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -840,7 +840,9 @@ class BeamModulePlugin implements Plugin<Project> {
         log4j2_log4j12_api                          : "org.apache.logging.log4j:log4j-1.2-api:$log4j2_version",
         mockito_core                                : "org.mockito:mockito-core:4.11.0",
         mockito_inline                              : "org.mockito:mockito-inline:4.11.0",
-        mongo_java_driver                           : "org.mongodb:mongo-java-driver:3.12.11",
+        mongo_java_driver                           : "org.mongodb:mongodb-driver-sync:5.5.0",
+        mongo_bson                                  : "org.mongodb:bson:5.5.0",
+        mongodb_driver_core                         : "org.mongodb:mongodb-driver-core:5.5.0",
         nemo_compiler_frontend_beam                 : "org.apache.nemo:nemo-compiler-frontend-beam:$nemo_version",
         netty_all                                   : "io.netty:netty-all:$netty_version",
         netty_handler                               : "io.netty:netty-handler:$netty_version",

it/mongodb/build.gradle

@@ -35,6 +35,7 @@ dependencies {
     implementation library.java.testcontainers_mongodb
     implementation library.java.google_code_gson
     implementation library.java.mongo_java_driver
+    implementation library.java.mongo_bson
     implementation library.java.vendored_guava_32_1_2_jre
 
     testImplementation library.java.mockito_core

sdks/java/extensions/sql/build.gradle

@@ -92,6 +92,9 @@ dependencies {
   implementation "org.codehaus.janino:commons-compiler:3.0.11"
   implementation library.java.jackson_core
   implementation library.java.mongo_java_driver
+  permitUnusedDeclared library.java.mongo_java_driver
+  implementation library.java.mongo_bson
+  implementation library.java.mongodb_driver_core
   implementation library.java.slf4j_api
   implementation library.java.joda_time
   implementation library.java.vendored_guava_32_1_2_jre
@@ -131,6 +134,7 @@ dependencies {
   testImplementation library.java.kafka_clients
   testImplementation project(":sdks:java:io:kafka")
   testImplementation project(path: ":sdks:java:io:mongodb", configuration: "testRuntimeMigration")
+  testImplementation library.java.mongo_java_driver
   testImplementation project(path: ":sdks:java:io:thrift", configuration: "testRuntimeMigration")
   testImplementation project(path: ":sdks:java:extensions:protobuf", configuration: "testRuntimeMigration")
   testCompileOnly project(":sdks:java:extensions:sql:udf-test-provider")

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/mongodb/MongoDbReadWriteIT.java

@@ -31,7 +31,8 @@
 import static org.hamcrest.core.IsInstanceOf.instanceOf;
 
 import com.mongodb.BasicDBObject;
-import com.mongodb.MongoClient;
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.client.model.Filters;
@@ -128,14 +129,14 @@ public static void setUp() throws Exception {
             .build();
     mongodExecutable = mongodStarter.prepare(mongodConfig);
     mongodProcess = mongodExecutable.start();
-    client = new MongoClient(hostname, port);
+    client = MongoClients.create("mongodb://" + hostname + ":" + port);
 
     mongoSqlUrl = String.format("mongodb://%s:%d/%s/%s", hostname, port, database, collection);
   }
 
   @AfterClass
   public static void tearDown() throws Exception {
-    client.dropDatabase(database);
+    client.getDatabase(database).drop();
     client.close();
     mongodProcess.stop();
     mongodExecutable.stop();

sdks/java/io/mongodb/build.gradle

@@ -28,13 +28,14 @@ dependencies {
   implementation project(path: ":sdks:java:core", configuration: "shadow")
   implementation library.java.joda_time
   implementation library.java.mongo_java_driver
+  implementation library.java.mongo_bson
+  implementation library.java.mongodb_driver_core
   implementation library.java.slf4j_api
   implementation library.java.vendored_guava_32_1_2_jre
   testImplementation library.java.junit
   testImplementation project(path: ":sdks:java:io:common")
   testImplementation project(path: ":sdks:java:testing:test-utils")
-  testImplementation "de.flapdoodle.embed:de.flapdoodle.embed.mongo:3.0.0"
-  testImplementation "de.flapdoodle.embed:de.flapdoodle.embed.process:3.0.0"
+  testImplementation "de.flapdoodle.embed:de.flapdoodle.embed.mongo:3.5.4"
   testRuntimeOnly library.java.slf4j_jdk14
   testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
 }

sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/FindQuery.java

@@ -21,7 +21,7 @@
 
 import com.google.auto.value.AutoValue;
 import com.mongodb.BasicDBObject;
-import com.mongodb.MongoClient;
+import com.mongodb.MongoClientSettings;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
 import com.mongodb.client.model.Projections;
@@ -79,7 +79,8 @@ private FindQuery withFilters(BsonDocument filters) {
 
   /** Convert the Bson filters into a BsonDocument via default encoding. */
   static BsonDocument bson2BsonDocument(Bson filters) {
-    return filters.toBsonDocument(BasicDBObject.class, MongoClient.getDefaultCodecRegistry());
+    return filters.toBsonDocument(
+        BasicDBObject.class, MongoClientSettings.getDefaultCodecRegistry());
   }
 
   /** Sets the filters to find. */