Distributed Transactions from the Java SDK

dependencies {
    compile group: 'com.couchbase.client', name: 'couchbase-transactions', version: '1.2.4'
}

<dependency>
    <groupId>com.couchbase.client</groupId>
    <artifactId>couchbase-transactions</artifactId>
    <version>1.2.4</version>
</dependency>

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

import com.couchbase.client.core.cnc.Event;
import com.couchbase.client.core.cnc.RequestSpan;
import com.couchbase.client.java.Bucket;
import com.couchbase.client.java.Cluster;
import com.couchbase.client.java.Collection;
import com.couchbase.client.java.Scope;
import com.couchbase.client.java.ReactiveCollection;
import com.couchbase.client.java.ReactiveScope;
import com.couchbase.client.java.Scope;
import com.couchbase.client.java.json.JsonArray;
import com.couchbase.client.java.json.JsonObject;
import com.couchbase.client.java.kv.GetResult;
import com.couchbase.client.java.query.QueryOptions;
import com.couchbase.client.java.query.QueryProfile;
import com.couchbase.client.java.query.QueryResult;
import com.couchbase.client.java.query.QueryScanConsistency;
import com.couchbase.client.tracing.opentelemetry.OpenTelemetryRequestSpan;
import com.couchbase.transactions.SingleQueryTransactionResult;
import com.couchbase.transactions.TransactionDurabilityLevel;
import com.couchbase.transactions.TransactionGetResult;
import com.couchbase.transactions.TransactionQueryOptions;
import com.couchbase.transactions.TransactionResult;
import com.couchbase.transactions.Transactions;
import com.couchbase.transactions.config.PerTransactionConfigBuilder;
import com.couchbase.transactions.config.SingleQueryTransactionConfigBuilder;
import com.couchbase.transactions.config.TransactionConfigBuilder;
import com.couchbase.transactions.deferred.TransactionSerializedContext;
import com.couchbase.transactions.error.TransactionCommitAmbiguous;
import com.couchbase.transactions.error.TransactionFailed;
import com.couchbase.transactions.log.IllegalDocumentState;
import com.couchbase.transactions.log.LogDefer;
import com.couchbase.transactions.log.TransactionCleanupAttempt;
import com.couchbase.transactions.log.TransactionCleanupEndRunEvent;

import io.opentelemetry.api.trace.Span;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

// Initialize the Couchbase cluster
Cluster cluster = Cluster.connect("localhost", "username", "password");
Bucket bucket = cluster.bucket("travel-sample");
Scope scope = bucket.scope("inventory");
Collection collection = scope.collection("airport");

// Create the single Transactions object
Transactions transactions = Transactions.create(cluster);

Transactions transactions = Transactions.create(cluster,
        TransactionConfigBuilder.create().durabilityLevel(TransactionDurabilityLevel.PERSIST_TO_MAJORITY)
                .logOnFailure(true, Event.Severity.WARN)
                .build());

try {
    transactions.run((ctx) -> {
        // 'ctx' is an AttemptContext, which permits getting, inserting,
        // removing and replacing documents, performing N1QL queries, and committing or
        // rolling back the transaction.

        // ... Your transaction logic here ...

        // This call is optional - if you leave it off, the transaction
        // will be committed anyway.
        ctx.commit();
    });
} catch (TransactionCommitAmbiguous e) {
    // The application will of course want to use its own logging rather
    // than System.err
    System.err.println("Transaction possibly committed");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
} catch (TransactionFailed e) {
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}

Mono<TransactionResult> result = transactions.reactive().run((ctx) -> {
    // 'ctx' is an AttemptContextReactive, providing asynchronous versions of the
    // AttemptContext methods

    return

    // Your transaction logic here: as an example, get and remove a doc
    ctx.get(collection.reactive(), "document-id").flatMap(doc -> ctx.remove(doc))

            // The commit call is optional - if you leave it off,
            // the transaction will be committed anyway.
            .then(ctx.commit());
}).doOnError(err -> {
    if (err instanceof TransactionCommitAmbiguous) {
        System.err.println("Transaction possibly committed: ");
    } else {
        System.err.println("Transaction failed: ");
    }

    for (LogDefer e : ((TransactionFailed) err).result().log().logs()) {
        // System.err is used for example, log failures to your own logging system
        System.err.println(err.toString());
    }
});

// Normally you will chain this result further and ultimately subscribe. For
// simplicity, here we just block on the result.
TransactionResult finalResult = result.block();

Scope inventory = cluster.bucket("travel-sample").scope("inventory");

try {
    TransactionResult result = transactions.run((ctx) -> {
        // Inserting a doc:
        ctx.insert(collection, "doc-a", JsonObject.create());

        // Getting documents:
        // Use ctx.getOptional if the document may or may not exist
        Optional<TransactionGetResult> docOpt = ctx.getOptional(collection, "doc-a");

        // Use ctx.get if the document should exist, and the transaction
        // will fail if it does not
        TransactionGetResult docA = ctx.get(collection, "doc-a");

        // Replacing a doc:
        TransactionGetResult docB = ctx.get(collection, "doc-b");
        JsonObject content = docB.contentAs(JsonObject.class);
        content.put("transactions", "are awesome");
        ctx.replace(docB, content);

        // Removing a doc:
        TransactionGetResult docC = ctx.get(collection, "doc-c");
        ctx.remove(docC);

        // Performing a SELECT N1QL query against a scope:
        QueryResult qr = ctx.query(inventory, "SELECT * FROM hotel WHERE country = $1",
                TransactionQueryOptions.queryOptions()
                        .parameters(JsonArray.from("United Kingdom")));
        List<JsonObject> rows = qr.rowsAs(JsonObject.class);

        // Performing an UPDATE N1QL query on multiple documents, in the `inventory` scope:
        ctx.query(inventory, "UPDATE route SET airlineid = $1 WHERE airline = $2",
                TransactionQueryOptions.queryOptions()
                        .parameters(JsonArray.from("airline_137", "AF")));

        // Committing (the ctx.commit() call is optional)
        ctx.commit();
    });
} catch (TransactionCommitAmbiguous e) {
    System.err.println("Transaction possibly committed");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
} catch (TransactionFailed e) {
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}

ReactiveScope inventory = cluster.bucket("travel-sample").scope("inventory").reactive();

Mono<TransactionResult> result = transactions.reactive().run((ctx) -> {
    return
    // Inserting a doc:
    ctx.insert(collection.reactive(), "doc-a", JsonObject.create())

            // Getting and replacing a doc:
            .then(ctx.get(collection.reactive(), "doc-b")).flatMap(docB -> {
                JsonObject content = docB.contentAs(JsonObject.class);
                content.put("transactions", "are awesome");
                return ctx.replace(docB, content);
            })

            // Getting and removing a doc:
            .then(ctx.get(collection.reactive(), "doc-c"))
                .flatMap(doc -> ctx.remove(doc))

            // Performing a SELECT N1QL query, in the `inventory` scope:
            .then(ctx.query(inventory, "SELECT * FROM hotel WHERE country = $1",
                    TransactionQueryOptions.queryOptions()
                            .parameters(JsonArray.from("United Kingdom"))))

            .doOnNext(queryResult -> queryResult.rowsAs(JsonObject.class)
                    .doOnNext(row -> {
                        // the application would do something with each row here
                    }))

            // Performing an UPDATE N1QL query on multiple documents, in the `inventory` scope:
            .then(ctx.query(inventory, "UPDATE route SET airlineid = $1 WHERE airline = $2",
                    TransactionQueryOptions.queryOptions()
                            .parameters(JsonArray.from("airline_137", "AF"))))

            .then(ctx.commit());

}).doOnError(err -> {
    if (err instanceof TransactionCommitAmbiguous) {
        System.err.println("Transaction possibly committed: ");
    } else {
        System.err.println("Transaction failed: ");
    }

    for (LogDefer e : ((TransactionFailed) err).result().log().logs()) {
        // System.err is used for example, log failures to your own logging system
        System.err.println(err.toString());
    }
});

// Normally you will chain this result further and ultimately subscribe.
// For simplicity, here we just block on the result.
result.block();

transactions.run((ctx) -> {
    String docId = "a-doc";

    Optional<TransactionGetResult> docOpt = ctx.getOptional(collection, docId);
    TransactionGetResult doc = ctx.get(collection, docId);
});

transactions.run((ctx) -> {
    String docId = "docId";

    ctx.insert(collection, docId, JsonObject.create());

    Optional<TransactionGetResult> doc = ctx.getOptional(collection, docId);

    assert (doc.isPresent());
});

Mono<TransactionResult> result = transactions.reactive().run((ctx) -> {
    return ctx.get(collection.reactive(), "anotherDoc").flatMap(doc -> {
        JsonObject content = doc.contentAs(JsonObject.class);
        content.put("transactions", "are awesome");
        return ctx.replace(doc, content);
    }).then();
});

public void playerHitsMonster(int damage, String playerId, String monsterId) {
    Transactions transactions = getTransactions();

    try {
        transactions.run((ctx) -> {
            TransactionGetResult monsterDoc = ctx.get(collection, monsterId);
            TransactionGetResult playerDoc = ctx.get(collection, playerId);

            int monsterHitpoints = monsterDoc.contentAs(JsonObject.class).getInt("hitpoints");
            int monsterNewHitpoints = monsterHitpoints - damage;

            if (monsterNewHitpoints <= 0) {
                // Monster is killed. The remove is just for demoing, and a more realistic
                // example would set a
                // "dead" flag or similar.
                ctx.remove(monsterDoc);

                // The player earns experience for killing the monster
                int experienceForKillingMonster = monsterDoc.contentAs(JsonObject.class)
                        .getInt("experienceWhenKilled");
                int playerExperience = playerDoc.contentAs(JsonObject.class).getInt("experience");
                int playerNewExperience = playerExperience + experienceForKillingMonster;
                int playerNewLevel = calculateLevelForExperience(playerNewExperience);

                JsonObject playerContent = playerDoc.contentAs(JsonObject.class);

                playerContent.put("experience", playerNewExperience);
                playerContent.put("level", playerNewLevel);

                ctx.replace(playerDoc, playerContent);
            } else {
                // Monster is damaged but still alive
                JsonObject monsterContent = monsterDoc.contentAs(JsonObject.class);
                monsterContent.put("hitpoints", monsterNewHitpoints);

                ctx.replace(monsterDoc, monsterContent);
            }
        });
    } catch (TransactionFailed e) {
        // The operation failed. Both the monster and the player will be untouched.

        // Situations that can cause this would include either the monster
        // or player not existing (as get is used), or a persistent
        // failure to be able to commit the transaction, for example on
        // prolonged node failure.
    }
}

cluster.environment().eventBus().subscribe(event -> {
    if (event instanceof IllegalDocumentState) {
        // log this event for review
    }
});

class BalanceInsufficient extends RuntimeException {
}

try {
    transactions.run((ctx) -> {
        TransactionGetResult customer = ctx.get(collection, "customer-name");

        if (customer.contentAsObject().getInt("balance") < costOfItem) {
            throw new BalanceInsufficient();
        }
        // else continue transaction
    });
} catch (TransactionCommitAmbiguous e) {
    // This exception can only be thrown at the commit point, after the
    // BalanceInsufficient logic has been passed, so there is no need to
    // check getCause here.
    System.err.println("Transaction possibly committed");
    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
} catch (TransactionFailed e) {
    if (e.getCause() instanceof BalanceInsufficient) {
        // Re-raise the error
        throw (RuntimeException) e.getCause();
    } else {
        System.err.println("Transaction did not reach commit point");

        for (LogDefer err : e.result().log().logs()) {
            System.err.println(err.toString());
        }
    }
}

transactions.run((ctx) -> {
    TransactionGetResult customer = ctx.get(collection, "customer-name");

    if (customer.contentAsObject().getInt("balance") < costOfItem) {
        ctx.rollback();
    }
    // else continue transaction
});

} catch (TransactionCommitAmbiguous e) {
    // The application will of course want to use its own logging rather
    // than System.err
    System.err.println("Transaction possibly committed");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
} catch (TransactionFailed e) {
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}

}).doOnError(err -> {
    if (err instanceof TransactionCommitAmbiguous) {
        System.err.println("Transaction possibly committed: ");
    } else {
        System.err.println("Transaction failed: ");
    }

    for (LogDefer e : ((TransactionFailed) err).result().log().logs()) {
        // System.err is used for example, log failures to your own logging system
        System.err.println(err.toString());
    }
});

.logOnFailure(true, Event.Severity.WARN)

final Logger LOGGER = Logger.getLogger("transactions");

try {
    TransactionResult result = transactions.run((ctx) -> {
        // ... transactional code here ...
    });
} catch (TransactionCommitAmbiguous err) {
    // The transaction may or may not have reached commit point
    LOGGER.info("Transaction returned TransactionCommitAmbiguous and" + " may have succeeded, logs:");
    err.result().log().logs().forEach(log -> LOGGER.info(log.toString()));
} catch (TransactionFailed err) {
    // The transaction definitely did not reach commit point
    LOGGER.info("Transaction failed with TransactionFailed, logs:");
    err.result().log().logs().forEach(log -> LOGGER.info(log.toString()));
}

List<String> docIds = Arrays.asList("doc1", "doc2", "doc3", "doc4", "doc5");
ReactiveCollection coll = collection.reactive();
int concurrency = 100; // This many operations will be in-flight at once

TransactionResult result = transactions.reactive((ctx) -> {
    return Flux.fromIterable(docIds)
            .parallel(concurrency)
            .runOn(Schedulers.boundedElastic())
            .concatMap(docId -> ctx.get(collection.reactive(), docId)
                    .flatMap(doc -> {
                        JsonObject content = doc.contentAsObject();
                        content.put("value", "updated");
                        return ctx.replace(doc, content);
                    }))
            .sequential()
            .then();
}).block();

try {
    TransactionResult result = transactions.run((ctx) -> {
        JsonObject initial = JsonObject.create().put("val", 1);
        ctx.insert(collection, "a-doc-id", initial);

        // Defer means don't do a commit right now. `serialized` in the result will be
        // present.
        ctx.defer();
    });

    // Available because ctx.defer() was called
    assert (result.serialized().isPresent());

    TransactionSerializedContext serialized = result.serialized().get();

    // This is going to store a serialized form of the transaction to pass around
    byte[] encoded = serialized.encodeAsBytes();

} catch (TransactionFailed e) {
    // System.err is used for example, log failures to your own logging system
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}

TransactionSerializedContext serialized = TransactionSerializedContext.createFrom(encoded);

try {
    TransactionResult result = transactions.commit(serialized);

} catch (TransactionFailed e) {
    // System.err is used for example, log failures to your own logging system
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}

TransactionSerializedContext serialized = TransactionSerializedContext.createFrom(encoded);

try {
    TransactionResult result = transactions.rollback(serialized);

} catch (TransactionFailed e) {
    // System.err is used for example, log failures to your own logging system
    System.err.println("Transaction did not reach commit point");

    for (LogDefer err : e.result().log().logs()) {
        System.err.println(err.toString());
    }
}