Java - MongoDB Multi-Document ACID Transactions

1
git clone git@github.com:mongodb-developer/java-quick-start.git

1
docker run --rm -d -p 27017:27017 -h $(hostname) --name mongo mongo:7.0.5 --replSet=RS && sleep 3 && docker exec mongo mongosh --quiet --eval "rs.initiate();"

1
mvn compile exec:java -Dexec.mainClass="com.mongodb.quickstart.transactions.ChangeStreams" -Dmongodb.uri="mongodb+srv://USERNAME:PASSWORD@cluster0-abcde.mongodb.net/test?w=majority"

1
mvn compile exec:java -Dexec.mainClass="com.mongodb.quickstart.transactions.Transactions" -Dmongodb.uri="mongodb+srv://USERNAME:PASSWORD@cluster0-abcde.mongodb.net/test?w=majority"

1
db.product.insertMany([
2
    { "_id" : "beer", "price" : NumberDecimal("3.75"), "stock" : NumberInt(5) },
3
    { "_id" : "wine", "price" : NumberDecimal("7.5"), "stock" : NumberInt(3) }
4
])

1
cursor = db.product.watch([{$match: {operationType: "update"}}]);
2
while (!cursor.isClosed()) {
3
  let next = cursor.tryNext()
4
  while (next !== null) {
5
    printjson(next);
6
    next = cursor.tryNext()
7
  }
8
}

1
RS [direct: primary] test> db.product.updateMany({}, {$mul: {price:0.8}})
2
{
3
  acknowledged: true,
4
  insertedId: null,
5
  matchedCount: 2,
6
  modifiedCount: 2,
7
  upsertedCount: 0
8
}
9
RS [direct: primary] test> db.product.find().pretty()
10
[
11
  { _id: 'beer', price: Decimal128("3.00000000000000000"), stock: 5 },
12
  { _id: 'wine', price: Decimal128("6.0000000000000000"), stock: 3 }
13
]

1
{
2
  _id: {
3
    _data: '8265580539000000012B042C0100296E5A1004A7F55A5B35BD4C7DB2CD56C6CFEA9C49463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B657900463C5F6964003C6265657200000004'
4
  },
5
  operationType: 'update',
6
  clusterTime: Timestamp({ t: 1700267321, i: 1 }),
7
  wallTime: ISODate("2023-11-18T00:28:41.601Z"),
8
  ns: {
9
    db: 'test',
10
    coll: 'product'
11
  },
12
  documentKey: {
13
    _id: 'beer'
14
  },
15
  updateDescription: {
16
    updatedFields: {
17
      price: Decimal128("3.00000000000000000")
18
    },
19
    removedFields: [],
20
    truncatedArrays: []
21
  }
22
}
23
{
24
  _id: {
25
    _data: '8265580539000000022B042C0100296E5A1004A7F55A5B35BD4C7DB2CD56C6CFEA9C49463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B657900463C5F6964003C77696E6500000004'
26
  },
27
  operationType: 'update',
28
  clusterTime: Timestamp({ t: 1700267321, i: 2 }),
29
  wallTime: ISODate("2023-11-18T00:28:41.601Z"),
30
  ns: {
31
    db: 'test',
32
    coll: 'product'
33
  },
34
  documentKey: {
35
    _id: 'wine'
36
  },
37
  updateDescription: {
38
    updatedFields: {
39
      price: Decimal128("6.0000000000000000")
40
    },
41
    removedFields: [],
42
    truncatedArrays: []
43
  }
44
}

1
RS [direct: primary] test> db.product.updateOne({_id: "wine"},{$inc: {stock:1}, $set: {description : "It's the best wine on Earth"}})
2
{
3
  acknowledged: true,
4
  insertedId: null,
5
  matchedCount: 1,
6
  modifiedCount: 1,
7
  upsertedCount: 0
8
}
9
RS [direct: primary] test> db.product.findOne({_id: "wine"})
10
{
11
  _id: 'wine',
12
  price: Decimal128("6.0000000000000000"),
13
  stock: 4,
14
  description: 'It's the best wine on Earth'
15
}

1
{ "_id" : "beer", "price" : NumberDecimal("3"), "stock" : NumberInt(5) }

1
{
2
  "_id" : "Alice",
3
  "items" : [
4
    {
5
      "price" : NumberDecimal("3"),
6
      "productId" : "beer",
7
      "quantity" : NumberInt(2)
8
    }
9
  ]
10
}

1
db.product.drop()
2
db.createCollection("product", {
3
   validator: {
4
      $jsonSchema: {
5
         bsonType: "object",
6
         required: [ "_id", "price", "stock" ],
7
         properties: {
8
            _id: {
9
               bsonType: "string",
10
               description: "must be a string and is required"
11
            },
12
            price: {
13
               bsonType: "decimal",
14
               minimum: 0,
15
               description: "must be a non-negative decimal and is required"
16
            },
17
            stock: {
18
               bsonType: "int",
19
               minimum: 0,
20
               description: "must be a non-negative integer and is required"
21
            }
22
         }
23
      }
24
   }
25
})

1
package com.mongodb.quickstart.transactions;
2
3
import com.mongodb.ConnectionString;
4
import com.mongodb.MongoClientSettings;
5
import com.mongodb.client.MongoClient;
6
import com.mongodb.client.MongoClients;
7
import com.mongodb.client.MongoDatabase;
8
import com.mongodb.client.model.CreateCollectionOptions;
9
import com.mongodb.client.model.ValidationAction;
10
import com.mongodb.client.model.ValidationOptions;
11
import org.bson.BsonDocument;
12
13
import static com.mongodb.client.model.changestream.FullDocument.UPDATE_LOOKUP;
14
15
public class ChangeStreams {
16
17
    private static final String CART = "cart";
18
    private static final String PRODUCT = "product";
19
20
    public static void main(String[] args) {
21
        ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
22
        MongoClientSettings clientSettings = MongoClientSettings.builder()
23
                                                                .applyConnectionString(connectionString)
24
                                                                .build();
25
        try (MongoClient client = MongoClients.create(clientSettings)) {
26
            MongoDatabase db = client.getDatabase("test");
27
            System.out.println("Dropping the '" + db.getName() + "' database.");
28
            db.drop();
29
            System.out.println("Creating the '" + CART + "'  collection.");
30
            db.createCollection(CART);
31
            System.out.println("Creating the '" + PRODUCT + "' collection with a JSON Schema.");
32
            db.createCollection(PRODUCT, productJsonSchemaValidator());
33
            System.out.println("Watching the collections in the DB " + db.getName() + "...");
34
            db.watch()
35
              .fullDocument(UPDATE_LOOKUP)
36
              .forEach(doc -> System.out.println(doc.getClusterTime() + " => " + doc.getFullDocument()));
37
        }
38
    }
39
40
    private static CreateCollectionOptions productJsonSchemaValidator() {
41
        String jsonSchema = """
42
                            {
43
                              "$jsonSchema": {
44
                                "bsonType": "object",
45
                                "required": ["_id", "price", "stock"],
46
                                "properties": {
47
                                  "_id": {
48
                                    "bsonType": "string",
49
                                    "description": "must be a string and is required"
50
                                  },
51
                                  "price": {
52
                                    "bsonType": "decimal",
53
                                    "minimum": 0,
54
                                    "description": "must be a non-negative decimal and is required"
55
                                  },
56
                                  "stock": {
57
                                    "bsonType": "int",
58
                                    "minimum": 0,
59
                                    "description": "must be a non-negative integer and is required"
60
                                  }
61
                                }
62
                              }
63
                            }""";
64
        return new CreateCollectionOptions().validationOptions(
65
                new ValidationOptions().validationAction(ValidationAction.ERROR)
66
                                       .validator(BsonDocument.parse(jsonSchema)));
67
    }
68
}

1
package com.mongodb.quickstart.transactions;
2
3
import com.mongodb.*;
4
import com.mongodb.client.*;
5
import com.mongodb.quickstart.transactions.models.Cart;
6
import com.mongodb.quickstart.transactions.models.Product;
7
import org.bson.BsonDocument;
8
import org.bson.codecs.configuration.CodecRegistry;
9
import org.bson.codecs.pojo.PojoCodecProvider;
10
import org.bson.conversions.Bson;
11
12
import java.math.BigDecimal;
13
import java.util.ArrayList;
14
import java.util.Collections;
15
import java.util.List;
16
17
import static com.mongodb.client.model.Filters.*;
18
import static com.mongodb.client.model.Updates.inc;
19
import static org.bson.codecs.configuration.CodecRegistries.fromProviders;
20
import static org.bson.codecs.configuration.CodecRegistries.fromRegistries;
21
22
public class Transactions {
23
24
    private static final BigDecimal BEER_PRICE = BigDecimal.valueOf(3);
25
    private static final String BEER_ID = "beer";
26
    private static final Bson filterId = eq("_id", BEER_ID);
27
    private static final Bson filterAlice = eq("_id", "Alice");
28
    private static final Bson matchBeer = elemMatch("items", eq("productId", "beer"));
29
    private static final Bson incrementTwoBeers = inc("items.$.quantity", 2);
30
    private static final Bson decrementTwoBeers = inc("stock", -2);
31
    private static MongoCollection<Cart> cartCollection;
32
    private static MongoCollection<Product> productCollection;
33
34
    public static void main(String[] args) {
35
        ConnectionString connectionString = new ConnectionString(System.getProperty("mongodb.uri"));
36
        CodecRegistry pojoCodecRegistry = fromProviders(PojoCodecProvider.builder().automatic(true).build());
37
        CodecRegistry codecRegistry = fromRegistries(MongoClientSettings.getDefaultCodecRegistry(), pojoCodecRegistry);
38
        MongoClientSettings clientSettings = MongoClientSettings.builder()
39
                                                                .applyConnectionString(connectionString)
40
                                                                .codecRegistry(codecRegistry)
41
                                                                .build();
42
        try (MongoClient client = MongoClients.create(clientSettings)) {
43
            MongoDatabase db = client.getDatabase("test");
44
            cartCollection = db.getCollection("cart", Cart.class);
45
            productCollection = db.getCollection("product", Product.class);
46
            transactionsDemo(client);
47
        }
48
    }
49
50
    private static void transactionsDemo(MongoClient client) {
51
        clearCollections();
52
        insertProductBeer();
53
        printDatabaseState();
54
        System.out.println("""
55
                           #########  NO  TRANSACTION #########
56
                           Alice wants 2 beers.
57
                           We have to create a cart in the 'cart' collection and update the stock in the 'product' collection.
58
                           The 2 actions are correlated but can not be executed at the same cluster time.
59
                           Any error blocking one operation could result in stock error or a sale of beer that we can't fulfill as we have no stock.
60
                           ------------------------------------""");
61
        aliceWantsTwoBeers();
62
        sleep();
63
        removingBeersFromStock();
64
        System.out.println("####################################\n");
65
        printDatabaseState();
66
        sleep();
67
        System.out.println("""
68
                           ######### WITH TRANSACTION #########
69
                           Alice wants 2 extra beers.
70
                           Now we can update the 2 collections simultaneously.
71
                           The 2 operations only happen when the transaction is committed.
72
                           ------------------------------------""");
73
        aliceWantsTwoExtraBeersInTransactionThenCommitOrRollback(client);
74
        sleep();
75
        System.out.println("""
76
                           ######### WITH TRANSACTION #########
77
                           Alice wants 2 extra beers.
78
                           This time we do not have enough beers in stock so the transaction will rollback.
79
                           ------------------------------------""");
80
        aliceWantsTwoExtraBeersInTransactionThenCommitOrRollback(client);
81
    }
82
83
    private static void aliceWantsTwoExtraBeersInTransactionThenCommitOrRollback(MongoClient client) {
84
        ClientSession session = client.startSession();
85
        try {
86
            session.startTransaction(TransactionOptions.builder().writeConcern(WriteConcern.MAJORITY).build());
87
            aliceWantsTwoExtraBeers(session);
88
            sleep();
89
            removingBeerFromStock(session);
90
            session.commitTransaction();
91
        } catch (MongoException e) {
92
            session.abortTransaction();
93
            System.out.println("####### ROLLBACK TRANSACTION #######");
94
        } finally {
95
            session.close();
96
            System.out.println("####################################\n");
97
            printDatabaseState();
98
        }
99
    }
100
101
    private static void removingBeersFromStock() {
102
        System.out.println("Trying to update beer stock : -2 beers.");
103
        try {
104
            productCollection.updateOne(filterId, decrementTwoBeers);
105
        } catch (MongoException e) {
106
            System.out.println("########   MongoException   ########");
107
            System.out.println("##### STOCK CANNOT BE NEGATIVE #####");
108
            throw e;
109
        }
110
    }
111
112
    private static void removingBeerFromStock(ClientSession session) {
113
        System.out.println("Trying to update beer stock : -2 beers.");
114
        try {
115
            productCollection.updateOne(session, filterId, decrementTwoBeers);
116
        } catch (MongoException e) {
117
            System.out.println("########   MongoException   ########");
118
            System.out.println("##### STOCK CANNOT BE NEGATIVE #####");
119
            throw e;
120
        }
121
    }
122
123
    private static void aliceWantsTwoBeers() {
124
        System.out.println("Alice adds 2 beers in her cart.");
125
        cartCollection.insertOne(new Cart("Alice", List.of(new Cart.Item(BEER_ID, 2, BEER_PRICE))));
126
    }
127
128
    private static void aliceWantsTwoExtraBeers(ClientSession session) {
129
        System.out.println("Updating Alice cart : adding 2 beers.");
130
        cartCollection.updateOne(session, and(filterAlice, matchBeer), incrementTwoBeers);
131
    }
132
133
    private static void insertProductBeer() {
134
        productCollection.insertOne(new Product(BEER_ID, 5, BEER_PRICE));
135
    }
136
137
    private static void clearCollections() {
138
        productCollection.deleteMany(new BsonDocument());
139
        cartCollection.deleteMany(new BsonDocument());
140
    }
141
142
    private static void printDatabaseState() {
143
        System.out.println("Database state:");
144
        printProducts(productCollection.find().into(new ArrayList<>()));
145
        printCarts(cartCollection.find().into(new ArrayList<>()));
146
        System.out.println();
147
    }
148
149
    private static void printProducts(List<Product> products) {
150
        products.forEach(System.out::println);
151
    }
152
153
    private static void printCarts(List<Cart> carts) {
154
        if (carts.isEmpty()) {
155
            System.out.println("No carts...");
156
        } else {
157
            carts.forEach(System.out::println);
158
        }
159
    }
160
161
    private static void sleep() {
162
        System.out.println("Sleeping 1 second...");
163
        try {
164
            Thread.sleep(1000);
165
        } catch (InterruptedException e) {
166
            System.err.println("Oops!");
167
            e.printStackTrace();
168
        }
169
    }
170
}

1
Dropping the 'test' database.
2
Creating the 'cart'  collection.
3
Creating the 'product' collection with a JSON Schema.
4
Watching the collections in the DB test...
5
Timestamp{value=7304460075832180737, seconds=1700702141, inc=1} => Document{{_id=beer, price=3, stock=5}}
6
Timestamp{value=7304460075832180738, seconds=1700702141, inc=2} => Document{{_id=Alice, items=[Document{{price=3, productId=beer, quantity=2}}]}}
7
Timestamp{value=7304460080127148033, seconds=1700702142, inc=1} => Document{{_id=beer, price=3, stock=3}}
8
Timestamp{value=7304460088717082625, seconds=1700702144, inc=1} => Document{{_id=Alice, items=[Document{{price=3, productId=beer, quantity=4}}]}}
9
Timestamp{value=7304460088717082625, seconds=1700702144, inc=1} => Document{{_id=beer, price=3, stock=1}}

1
Database state:
2
Product{id='beer', stock=5, price=3}
3
No carts...
4
5
#########  NO  TRANSACTION #########
6
Alice wants 2 beers.
7
We have to create a cart in the 'cart' collection and update the stock in the 'product' collection.
8
The 2 actions are correlated but can not be executed on the same cluster time.
9
Any error blocking one operation could result in stock error or a sale of beer that we can't fulfill as we have no stock.
10
------------------------------------
11
Alice adds 2 beers in her cart.
12
Sleeping 1 second...
13
Trying to update beer stock : -2 beers.
14
####################################
15
16
Database state:
17
Product{id='beer', stock=3, price=3}
18
Cart{id='Alice', items=[Item{productId=beer, quantity=2, price=3}]}
19
20
Sleeping 1 second...
21
######### WITH TRANSACTION #########
22
Alice wants 2 extra beers.
23
Now we can update the 2 collections simultaneously.
24
The 2 operations only happen when the transaction is committed.
25
------------------------------------
26
Updating Alice cart : adding 2 beers.
27
Sleeping 1 second...
28
Trying to update beer stock : -2 beers.
29
####################################
30
31
Database state:
32
Product{id='beer', stock=1, price=3}
33
Cart{id='Alice', items=[Item{productId=beer, quantity=4, price=3}]}
34
35
Sleeping 1 second...
36
######### WITH TRANSACTION #########
37
Alice wants 2 extra beers.
38
This time we do not have enough beers in stock so the transaction will rollback.
39
------------------------------------
40
Updating Alice cart : adding 2 beers.
41
Sleeping 1 second...
42
Trying to update beer stock : -2 beers.
43
########   MongoException   ########
44
##### STOCK CANNOT BE NEGATIVE #####
45
####### ROLLBACK TRANSACTION #######
46
####################################
47
48
Database state:
49
Product{id='beer', stock=1, price=3}
50
Cart{id='Alice', items=[Item{productId=beer, quantity=4, price=3}]}