@Override
  public BeamSqlTable buildBeamSqlTable(Table table) {
    return new MongoDbTable(table);
  }
@Test
  public void testBuildBeamSqlTable_withBadLocation_throwsException() {
    ImmutableList<String> badLocations =
        ImmutableList.of(
            "mongodb://localhost:27017/database/",
            "mongodb://localhost:27017/database",
            "localhost:27017/database/collection",
            "mongodb://:27017/database/collection",
            "mongodb://localhost:27017//collection",
            "mongodb://localhost/database/collection",
            "mongodb://localhost:/database/collection");

    for (String badLocation : badLocations) {
      Table table = fakeTable("TEST", badLocation);
      assertThrows(IllegalArgumentException.class, () -> provider.buildBeamSqlTable(table));
    }
  }