InventoryUpdateWithTransaction.java

package io.gitrebase;

import com.mongodb.ReadConcern;
import com.mongodb.ReadPreference;
import com.mongodb.TransactionOptions;
import com.mongodb.WriteConcern;
import com.mongodb.client.*;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Updates;
import org.bson.Document;

public class InventoryUpdateWithTransaction {

    private static final String DATABASE_NAME = "gitrebase";
    private static final String COLLECTION_PRODUCTS = "products";
    private static final String COLLECTION_ORDERS = "orders";
    private static final String MONGO_URI = "mongodb://localhost:27017,localhost:27018,localhost:27019/?replicaSet=rs0";

    public static void main(String[] args) throws InterruptedException {
        MongoClient client = MongoClients.create(MONGO_URI);
        MongoDatabase database = client.getDatabase(DATABASE_NAME);
        MongoCollection<Document> products = database.getCollection(COLLECTION_PRODUCTS);
        MongoCollection<Document> orders = database.getCollection(COLLECTION_ORDERS);

        products.deleteMany(Filters.eq("category", "PIZZA"));
        Document pizza = new Document("_id", "PIZZA_001")
                .append("name", "Cheese Burst Pizza")
                .append("category", "PIZZA")
                .append("price", 350);
        products.insertOne(pizza);
        System.out.println("Inserted product: " + pizza.toJson());

        // Write to the same collection from another thread
        Thread clientAThread = new Thread(() -> {

            final ClientSession clientSession = client.startSession();
            TransactionOptions txnOptions = TransactionOptions.builder()
                    .readPreference(ReadPreference.primary())
                    .readConcern(ReadConcern.SNAPSHOT)
                    .writeConcern(WriteConcern.MAJORITY)
                    .build();
            TransactionBody txnBody = (TransactionBody<String>) () -> {
                try {
                    System.out.println("Client A: Fetching product ...");
                    Document firstRead = products.find(clientSession, Filters.eq("_id", "PIZZA_001")).first();
                    System.out.println("Client A : " + firstRead.toJson());
                    Thread.sleep(3000);
                    System.out.println("Client A: Placing order ...");
                    orders.insertOne(new Document("orderId", "ORD_001")
                            .append("productId", "PIZZA_001"));
                    System.out.println("Client A: Order placed ");
                    Thread.sleep(1000);
                    System.out.println("Client A: Fetching product ...");
                    Document secondRead = products.find(clientSession, Filters.eq("_id", "PIZZA_001")).first();
                    System.out.println("Client A : " + secondRead.toJson());
                    return "Inserted into collections in different databases";
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            };
            try {
                clientSession.withTransaction(txnBody, txnOptions);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                clientSession.close();
            }
        });


        // Write to the same collection from another thread
        Thread clientBThread = new Thread(() -> {
            try {
                Thread.sleep(1000);
                // Increment pizza price by 10%
                System.out.println("Client B: Incrementing pizza price by 10% ...");
                products.updateMany(Filters.eq("category", "PIZZA"), Updates.mul("price", 1.10));
                System.out.println("Client B: Price updated.");
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        clientBThread.start();
        clientAThread.start();

        clientAThread.join();
        clientBThread.join();
    }
}