Sessions and Transactions

/**
 * Client session interface for transactions and causally consistent operations
 */
public interface ClientSession extends AutoCloseable {
    /**
     * Gets the server address this session is pinned to (if any)
     * @return server address or null if not pinned
     */
    ServerAddress getPinnedServerAddress();
    
    /**
     * Checks if this session has an active transaction
     * @return true if transaction is active
     */
    boolean hasActiveTransaction();
    
    /**
     * Gets the current transaction options
     * @return transaction options or null if no active transaction
     */
    TransactionOptions getTransactionOptions();
    
    /**
     * Starts a new transaction with default options
     */
    void startTransaction();
    
    /**
     * Starts a new transaction with specific options
     * @param transactionOptions configuration for the transaction
     */
    void startTransaction(TransactionOptions transactionOptions);
    
    /**
     * Commits the current transaction
     */
    void commitTransaction();
    
    /**
     * Aborts the current transaction
     */
    void abortTransaction();
    
    /**
     * Executes a transaction body with automatic retry logic
     * @param transactionBody the operation to execute in the transaction
     * @return result from the transaction body
     */
    <T> T withTransaction(TransactionBody<T> transactionBody);
    
    /**
     * Executes a transaction body with specific callback options
     * @param transactionBody the operation to execute in the transaction
     * @param options configuration for transaction callback behavior
     * @return result from the transaction body
     */
    <T> T withTransaction(TransactionBody<T> transactionBody, TransactionOptions options);
    
    /**
     * Notifies the session that a message has been sent (internal use)
     * @return true if the session should be considered active
     */
    boolean notifyMessageSent();
    
    /**
     * Notifies the session that an operation has been initiated (internal use)
     * @param operation the operation being initiated
     * @return true if the operation should proceed
     */
    boolean notifyOperationInitiated(Object operation);
    
    /**
     * Closes the session and releases server resources
     */
    void close();
}

import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.TransactionOptions;
import com.mongodb.ReadConcern;
import com.mongodb.WriteConcern;

// Basic session usage
try (MongoClient client = MongoClients.create()) {
    try (ClientSession session = client.startSession()) {
        MongoDatabase database = client.getDatabase("mydb");
        MongoCollection<Document> collection = database.getCollection("accounts");
        
        // Perform operations with session for causally consistent reads
        Document account = collection.find(session, Filters.eq("accountId", "12345")).first();
        collection.updateOne(session, 
            Filters.eq("accountId", "12345"), 
            Updates.inc("balance", 100));
    }
}

// Manual transaction control
try (ClientSession session = client.startSession()) {
    try {
        session.startTransaction();
        
        // Transfer money between accounts
        collection.updateOne(session,
            Filters.eq("accountId", "account1"),
            Updates.inc("balance", -100));
            
        collection.updateOne(session,
            Filters.eq("accountId", "account2"), 
            Updates.inc("balance", 100));
        
        session.commitTransaction();
        System.out.println("Transfer completed successfully");
        
    } catch (Exception e) {
        session.abortTransaction();
        System.err.println("Transfer failed, transaction aborted: " + e.getMessage());
        throw e;
    }
}

/**
 * Functional interface for transaction body execution
 * Transaction bodies must be idempotent as they may be retried
 */
@FunctionalInterface
public interface TransactionBody<T> {
    /**
     * Executes the transaction logic
     * This method must be idempotent as it may be retried on transient failures
     * @return result of the transaction
     * @throws Exception if the transaction should be aborted
     */
    T execute() throws Exception;
}

// Automatic transaction with retry logic
try (ClientSession session = client.startSession()) {
    String result = session.withTransaction(() -> {
        // This lambda must be idempotent
        MongoDatabase database = client.getDatabase("ecommerce");
        MongoCollection<Document> orders = database.getCollection("orders");
        MongoCollection<Document> inventory = database.getCollection("inventory");
        
        // Create order
        Document order = new Document("orderId", UUID.randomUUID().toString())
            .append("customerId", "customer123")
            .append("items", Arrays.asList(
                new Document("productId", "product456").append("quantity", 2)
            ))
            .append("status", "pending")
            .append("createdAt", new Date());
        
        orders.insertOne(session, order);
        
        // Update inventory
        UpdateResult inventoryUpdate = inventory.updateOne(session,
            Filters.eq("productId", "product456"),
            Updates.inc("stock", -2));
        
        if (inventoryUpdate.getMatchedCount() == 0) {
            throw new IllegalStateException("Product not found in inventory");
        }
        
        // Return order ID
        return order.getString("orderId");
    });
    
    System.out.println("Order created with ID: " + result);
}

// Transaction with specific options
TransactionOptions txnOptions = TransactionOptions.builder()
    .readConcern(ReadConcern.SNAPSHOT)
    .writeConcern(WriteConcern.MAJORITY)
    .maxCommitTime(30L, TimeUnit.SECONDS)
    .build();

try (ClientSession session = client.startSession()) {
    Boolean transferResult = session.withTransaction(() -> {
        // Complex business logic
        return performAccountTransfer(session, "from123", "to456", 250.00);
    }, txnOptions);
    
    if (transferResult) {
        System.out.println("Transfer completed successfully");
    }
}

/**
 * Transaction configuration options
 */
public class TransactionOptions {
    /**
     * Read concern for transaction operations
     */
    private ReadConcern readConcern;
    
    /**
     * Write concern for transaction commit
     */
    private WriteConcern writeConcern;
    
    /**
     * Read preference for transaction operations
     */
    private ReadPreference readPreference;
    
    /**
     * Maximum time for commit operation
     */
    private Long maxCommitTimeMS;
    
    // Builder pattern for configuration
    public static Builder builder() { return new Builder(); }
    
    public static class Builder {
        public Builder readConcern(ReadConcern readConcern);
        public Builder writeConcern(WriteConcern writeConcern);
        public Builder readPreference(ReadPreference readPreference);
        public Builder maxCommitTime(long maxCommitTime, TimeUnit timeUnit);
        public TransactionOptions build();
    }
}

// Configure transaction for strong consistency
TransactionOptions strongConsistency = TransactionOptions.builder()
    .readConcern(ReadConcern.SNAPSHOT)
    .writeConcern(WriteConcern.MAJORITY)
    .readPreference(ReadPreference.primary())
    .maxCommitTime(60, TimeUnit.SECONDS)
    .build();

// Configure transaction for performance
TransactionOptions performance = TransactionOptions.builder()
    .readConcern(ReadConcern.LOCAL)
    .writeConcern(WriteConcern.W1)
    .readPreference(ReadPreference.primaryPreferred())
    .maxCommitTime(10, TimeUnit.SECONDS)
    .build();

// Use configuration in session
try (ClientSession session = client.startSession()) {
    session.startTransaction(strongConsistency);
    
    // Perform critical operations requiring strong consistency
    performCriticalUpdate(session);
    
    session.commitTransaction();
}

// Session options configuration
ClientSessionOptions sessionOptions = ClientSessionOptions.builder()
    .causallyConsistent(true)
    .defaultTransactionOptions(TransactionOptions.builder()
        .readConcern(ReadConcern.MAJORITY)
        .writeConcern(WriteConcern.MAJORITY)
        .build())
    .build();

try (ClientSession session = client.startSession(sessionOptions)) {
    // Session operations
}

// Long-running session with multiple transactions
try (ClientSession session = client.startSession()) {
    // First transaction
    session.withTransaction(() -> {
        // Batch of related operations
        processBatchOfOrders(session, orderBatch1);
        return null;
    });
    
    // Some non-transactional work
    generateReports(session);
    
    // Second transaction
    session.withTransaction(() -> {
        // Another batch of operations
        processBatchOfOrders(session, orderBatch2);
        return null;
    });
}

// Session with custom retry logic
private <T> T executeWithCustomRetry(ClientSession session, TransactionBody<T> body) {
    int maxRetries = 3;
    int attempts = 0;
    
    while (attempts < maxRetries) {
        try {
            return session.withTransaction(body);
        } catch (MongoException e) {
            attempts++;
            if (attempts >= maxRetries || !isRetryableError(e)) {
                throw e;
            }
            // Wait before retry
            try {
                Thread.sleep(100 * attempts);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(ie);
            }
        }
    }
    throw new RuntimeException("Max retries exceeded");
}

try (ClientSession session = client.startSession()) {
    session.withTransaction(() -> {
        try {
            // Business logic that might fail
            processPayment(session, paymentInfo);
            updateOrderStatus(session, orderId, "paid");
            sendConfirmationEmail(customerEmail);
            
            return "success";
            
        } catch (PaymentDeclinedException e) {
            // Business logic error - should abort transaction
            updateOrderStatus(session, orderId, "payment_failed");
            throw new RuntimeException("Payment declined: " + e.getMessage());
            
        } catch (EmailServiceException e) {
            // Email failure shouldn't abort financial transaction
            // Log error but continue
            logger.warn("Failed to send confirmation email", e);
            return "success_no_email";
            
        } catch (MongoWriteException e) {
            // Database constraint violation
            logger.error("Database constraint violation", e);
            throw e; // Will cause transaction retry or abort
        }
    });
}

// Handle specific transaction errors
try {
    performTransactionalOperation(session);
} catch (MongoTransactionException e) {
    if (e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL)) {
        // Retry the entire transaction
        logger.info("Transient transaction error, retrying...");
        // Retry logic here
    } else if (e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL)) {
        // Check if transaction actually committed
        logger.warn("Unknown transaction commit result");
        // Verification logic here
    } else {
        // Permanent failure
        logger.error("Transaction failed permanently", e);
        throw e;
    }
}

// Keep transactions short and focused
session.withTransaction(() -> {
    // Good: Quick, focused operations
    collection.insertOne(session, document);
    relatedCollection.updateOne(session, filter, update);
    return null;
});

// Avoid in transactions: Long-running operations
session.withTransaction(() -> {
    // Bad: Don't do this in transactions
    // - Large bulk operations
    // - Network calls to external services  
    // - Complex aggregations
    // - File I/O operations
    
    // Better: Do these outside the transaction
    return null;
});

// Batch related operations efficiently
List<WriteModel<Document>> operations = Arrays.asList(
    new InsertOneModel<>(doc1),
    new UpdateOneModel<>(filter1, update1),
    new DeleteOneModel<>(filter2)
);

session.withTransaction(() -> {
    // Efficient: Single bulk operation
    collection.bulkWrite(session, operations);
    return null;
});

// Use appropriate read/write concerns
TransactionOptions fastTxn = TransactionOptions.builder()
    .readConcern(ReadConcern.LOCAL)        // Faster reads
    .writeConcern(WriteConcern.W1)         // Faster writes
    .maxCommitTime(5, TimeUnit.SECONDS)    // Quick timeout
    .build();

TransactionOptions safeTxn = TransactionOptions.builder()
    .readConcern(ReadConcern.SNAPSHOT)     // Consistent reads
    .writeConcern(WriteConcern.MAJORITY)   // Durable writes
    .maxCommitTime(30, TimeUnit.SECONDS)   // Longer timeout
    .build();