Distributed Transactions from the Java SDK

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

<dependency>
    <groupId>com.couchbase.client</groupId>
    <artifactId>couchbase-transactions</artifactId>
    <version>1.1.8</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.ReactiveCollection;
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.tracing.opentelemetry.OpenTelemetryRequestSpan;
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.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;

// Initialize the Couchbase cluster
Cluster cluster = Cluster.connect("localhost", "Administrator", "password");
Bucket bucket = cluster.bucket("travel-sample");
Collection collection = bucket.defaultCollection();

// 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, along with committing and
        // 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();

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);

        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());
    }
}

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))

            // Committing:
            .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());
});

transactions.run((ctx) -> {
    ctx.query("INSERT INTO `default` VALUES ('doc', {'hello':'world'})");

    String st = "SELECT `default`.* FROM `default` WHERE META().id = 'doc'";
    QueryResult qr = ctx.query(st);
    qr.rowsAsObject().forEach(row -> {
        System.out.println(row);
    });
});

transactions.run((ctx) -> {
    ctx.insert(collection, "doc", JsonObject.create().put("hello", "world"));

    String st = "SELECT `default`.* FROM `default` WHERE META().id = 'doc'";
    QueryResult qr = ctx.query(st);
    qr.rowsAsObject().forEach(row -> {
        System.out.println(row);
    });
});

transactions.run((ctx) -> {
    ctx.query("INSERT INTO `default` VALUES ('doc', {'hello':'world'})",
            TransactionQueryOptions.queryOptions().profile(QueryProfile.TIMINGS));
});

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();

TransactionResult result = transactions.reactive((ctx) -> {

    // Tracks whether all operations were successful
    AtomicBoolean allOpsSucceeded = new AtomicBoolean(true);

    // The first mutation must be done in serial, as it also creates a metadata
    // entry
    return ctx.get(coll, docIds.get(0)).flatMap(doc -> {
        JsonObject content = doc.contentAsObject();
        content.put("value", "updated");
        return ctx.replace(doc, content);
    })

            // Do all other docs in parallel
            .thenMany(Flux.fromIterable(docIds.subList(1, docIds.size()))
                    .flatMap(docId -> ctx.get(coll, docId).flatMap(doc -> {
                        JsonObject content = doc.contentAsObject();
                        content.put("value", "updated");
                        return ctx.replace(doc, content);
                    }).onErrorResume(err -> {
                        allOpsSucceeded.set(false);
                        // App should replace this with logging
                        err.printStackTrace();

                        // Allow other ops to finish
                        return Mono.empty();
                    }),

                            // Run these in parallel
                            docIds.size())

    // The commit or rollback must also be done in serial
    ).then(Mono.defer(() -> {
        // Commit iff all ops succeeded
        if (allOpsSucceeded.get()) {
            return ctx.commit();
        } else {
            throw new RuntimeException("Retry the transaction");
        }
    }));
}).block();

Transactions transactions = Transactions.create(cluster,
        TransactionConfigBuilder.create().metadataCollection(metadataCollection));

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());
    }
}