Repository: incubator-beam Updated Branches: refs/heads/master 7d069a65b -> 8bfa08519
[BEAM-856] Use free available port to start the test MongoDB instance [BEAM-856] Use random port for MongoDB instance in test, fix MongoClient init in estimatedSizeBytes and splitIntoBundles methods using provided URI Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/dacd5233 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/dacd5233 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/dacd5233 Branch: refs/heads/master Commit: dacd52334667a38955332ec03feee8079aeae571 Parents: 7d069a6 Author: Jean-Baptiste Onofré <[email protected]> Authored: Mon Oct 31 17:27:15 2016 +0100 Committer: Davor Bonaci <[email protected]> Committed: Sat Nov 12 00:12:18 2016 -0800 ---------------------------------------------------------------------- .../apache/beam/sdk/io/mongodb/MongoDbIO.java | 4 +-- .../sdk/io/mongodb/MongoDBGridFSIOTest.java | 28 ++++++++++----- .../beam/sdk/io/mongodb/MongoDbIOTest.java | 38 ++++++++++++++------ 3 files changed, 49 insertions(+), 21 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dacd5233/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java index 20b9265..2729602 100644 --- a/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java +++ b/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java @@ -216,7 +216,7 @@ public class MongoDbIO { @Override public long getEstimatedSizeBytes(PipelineOptions pipelineOptions) { - MongoClient mongoClient = new MongoClient(); + MongoClient mongoClient = new MongoClient(new MongoClientURI(spec.uri())); MongoDatabase mongoDatabase = mongoClient.getDatabase(spec.database()); // get the Mongo collStats object @@ -230,7 +230,7 @@ public class MongoDbIO { @Override public List<BoundedSource<Document>> splitIntoBundles(long desiredBundleSizeBytes, PipelineOptions options) { - MongoClient mongoClient = new MongoClient(); + MongoClient mongoClient = new MongoClient(new MongoClientURI(spec.uri())); MongoDatabase mongoDatabase = mongoClient.getDatabase(spec.database()); List<Document> splitKeys; http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dacd5233/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDBGridFSIOTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDBGridFSIOTest.java b/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDBGridFSIOTest.java index 5f4d122..dc27ee2 100644 --- a/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDBGridFSIOTest.java +++ b/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDBGridFSIOTest.java @@ -26,6 +26,7 @@ import com.mongodb.gridfs.GridFSDBFile; import com.mongodb.gridfs.GridFSInputFile; import de.flapdoodle.embed.mongo.MongodExecutable; +import de.flapdoodle.embed.mongo.MongodProcess; import de.flapdoodle.embed.mongo.MongodStarter; import de.flapdoodle.embed.mongo.config.IMongodConfig; import de.flapdoodle.embed.mongo.config.MongoCmdOptionsBuilder; @@ -45,6 +46,7 @@ import java.io.InputStreamReader; import java.io.OutputStream; import java.io.OutputStreamWriter; import java.io.Serializable; +import java.net.ServerSocket; import java.util.List; import java.util.Random; import java.util.Scanner; @@ -82,14 +84,21 @@ public class MongoDBGridFSIOTest implements Serializable { private static final Logger LOGGER = LoggerFactory.getLogger(MongoDBGridFSIOTest.class); private static final String MONGODB_LOCATION = "target/mongodb"; - private static final int PORT = 27017; private static final String DATABASE = "gridfs"; + private static final transient MongodStarter mongodStarter = MongodStarter.getDefaultInstance(); + private static transient MongodExecutable mongodExecutable; + private static transient MongodProcess mongodProcess; + + private static int port; @BeforeClass public static void setup() throws Exception { - LOGGER.info("Starting MongoDB embedded instance"); + try (ServerSocket serverSocket = new ServerSocket(0)) { + port = serverSocket.getLocalPort(); + } + LOGGER.info("Starting MongoDB embedded instance on {}", port); try { Files.forceDelete(new File(MONGODB_LOCATION)); } catch (Exception e) { @@ -100,7 +109,7 @@ public class MongoDBGridFSIOTest implements Serializable { .version(Version.Main.PRODUCTION) .configServer(false) .replication(new Storage(MONGODB_LOCATION, null, 0)) - .net(new Net("localhost", PORT, Network.localhostIsIPv6())) + .net(new Net("localhost", port, Network.localhostIsIPv6())) .cmdOptions(new MongoCmdOptionsBuilder() .syncDelay(10) .useNoPrealloc(true) @@ -108,12 +117,12 @@ public class MongoDBGridFSIOTest implements Serializable { .useNoJournal(true) .build()) .build(); - mongodExecutable = MongodStarter.getDefaultInstance().prepare(mongodConfig); - mongodExecutable.start(); + mongodExecutable = mongodStarter.prepare(mongodConfig); + mongodProcess = mongodExecutable.start(); LOGGER.info("Insert test data"); - Mongo client = new Mongo("localhost", PORT); + Mongo client = new Mongo("localhost", port); DB database = client.getDB(DATABASE); GridFS gridfs = new GridFS(database); @@ -159,6 +168,7 @@ public class MongoDBGridFSIOTest implements Serializable { @AfterClass public static void stop() throws Exception { LOGGER.info("Stopping MongoDB instance"); + mongodProcess.stop(); mongodExecutable.stop(); } @@ -169,7 +179,7 @@ public class MongoDBGridFSIOTest implements Serializable { PCollection<String> output = pipeline.apply( MongoDbGridFSIO.<String>read() - .withUri("mongodb://localhost:" + PORT) + .withUri("mongodb://localhost:" + port) .withDatabase(DATABASE)); PAssert.thatSingleton( @@ -199,7 +209,7 @@ public class MongoDBGridFSIOTest implements Serializable { PCollection<KV<String, Integer>> output = pipeline.apply( MongoDbGridFSIO.<KV<String, Integer>>read() - .withUri("mongodb://localhost:" + PORT) + .withUri("mongodb://localhost:" + port) .withDatabase(DATABASE) .withBucket("mapBucket") .withParser(new MongoDbGridFSIO.Parser<KV<String, Integer>>() { @@ -246,7 +256,7 @@ public class MongoDBGridFSIOTest implements Serializable { public void testSplit() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); MongoDbGridFSIO.Read<String> read = MongoDbGridFSIO.<String>read() - .withUri("mongodb://localhost:" + PORT) + .withUri("mongodb://localhost:" + port) .withDatabase(DATABASE); BoundedGridFSSource src = new BoundedGridFSSource(read, null); http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/dacd5233/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java ---------------------------------------------------------------------- diff --git a/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java b/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java index 9a88267..129e81c 100644 --- a/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java +++ b/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java @@ -25,6 +25,7 @@ import com.mongodb.client.MongoCursor; import com.mongodb.client.MongoDatabase; import de.flapdoodle.embed.mongo.MongodExecutable; +import de.flapdoodle.embed.mongo.MongodProcess; import de.flapdoodle.embed.mongo.MongodStarter; import de.flapdoodle.embed.mongo.config.IMongodConfig; import de.flapdoodle.embed.mongo.config.MongoCmdOptionsBuilder; @@ -37,6 +38,7 @@ import de.flapdoodle.embed.process.runtime.Network; import java.io.File; import java.io.Serializable; +import java.net.ServerSocket; import java.util.ArrayList; import org.apache.beam.sdk.testing.NeedsRunner; @@ -54,6 +56,7 @@ import org.bson.Document; import org.junit.After; import org.junit.Assert; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import org.junit.experimental.categories.Category; import org.slf4j.Logger; @@ -67,15 +70,29 @@ public class MongoDbIOTest implements Serializable { private static final Logger LOGGER = LoggerFactory.getLogger(MongoDbIOTest.class); private static final String MONGODB_LOCATION = "target/mongodb"; - private static final int PORT = 27017; private static final String DATABASE = "beam"; private static final String COLLECTION = "test"; + private static final MongodStarter mongodStarter = MongodStarter.getDefaultInstance(); + private transient MongodExecutable mongodExecutable; + private transient MongodProcess mongodProcess; + + private static int port; + + /** + * Looking for an available network port. + */ + @BeforeClass + public static void availablePort() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + port = serverSocket.getLocalPort(); + } + } @Before public void setup() throws Exception { - LOGGER.info("Starting MongoDB embedded instance"); + LOGGER.info("Starting MongoDB embedded instance on {}", port); try { Files.forceDelete(new File(MONGODB_LOCATION)); } catch (Exception e) { @@ -86,7 +103,7 @@ public class MongoDbIOTest implements Serializable { .version(Version.Main.PRODUCTION) .configServer(false) .replication(new Storage(MONGODB_LOCATION, null, 0)) - .net(new Net("localhost", PORT, Network.localhostIsIPv6())) + .net(new Net("localhost", port, Network.localhostIsIPv6())) .cmdOptions(new MongoCmdOptionsBuilder() .syncDelay(10) .useNoPrealloc(true) @@ -94,12 +111,12 @@ public class MongoDbIOTest implements Serializable { .useNoJournal(true) .build()) .build(); - mongodExecutable = MongodStarter.getDefaultInstance().prepare(mongodConfig); - mongodExecutable.start(); + mongodExecutable = mongodStarter.prepare(mongodConfig); + mongodProcess = mongodExecutable.start(); LOGGER.info("Insert test data"); - MongoClient client = new MongoClient("localhost", PORT); + MongoClient client = new MongoClient("localhost", port); MongoDatabase database = client.getDatabase(DATABASE); MongoCollection collection = database.getCollection(COLLECTION); @@ -119,6 +136,7 @@ public class MongoDbIOTest implements Serializable { @After public void stop() throws Exception { LOGGER.info("Stopping MongoDB instance"); + mongodProcess.stop(); mongodExecutable.stop(); } @@ -129,7 +147,7 @@ public class MongoDbIOTest implements Serializable { PCollection<Document> output = pipeline.apply( MongoDbIO.read() - .withUri("mongodb://localhost:" + PORT) + .withUri("mongodb://localhost:" + port) .withDatabase(DATABASE) .withCollection(COLLECTION)); @@ -163,7 +181,7 @@ public class MongoDbIOTest implements Serializable { PCollection<Document> output = pipeline.apply( MongoDbIO.read() - .withUri("mongodb://localhost:" + PORT) + .withUri("mongodb://localhost:" + port) .withDatabase(DATABASE) .withCollection(COLLECTION) .withFilter("{\"scientist\":\"Einstein\"}")); @@ -184,12 +202,12 @@ public class MongoDbIOTest implements Serializable { data.add(Document.parse(String.format("{\"scientist\":\"Test %s\"}", i))); } pipeline.apply(Create.of(data)) - .apply(MongoDbIO.write().withUri("mongodb://localhost:" + PORT).withDatabase("test") + .apply(MongoDbIO.write().withUri("mongodb://localhost:" + port).withDatabase("test") .withCollection("test")); pipeline.run(); - MongoClient client = new MongoClient("localhost", PORT); + MongoClient client = new MongoClient("localhost", port); MongoDatabase database = client.getDatabase("test"); MongoCollection collection = database.getCollection("test");
