Transactions

Transactions require Java client 8.0.0 and later, Aerospike Database 8.0.0 and later, and a strong-consistency namespace. See Create and use transactions for the full transaction lifecycle and status codes, and Failed commits for the commit failure scenarios and recommended handling shown below.

Caution

Any exception during the transaction body or a definite commit failure aborts and rolls back all changes in the transaction. If a recommit after an in-doubt commit also comes back in-doubt, the outcome is unknown: do not assume failure and retry with a new transaction, since the original transaction may have already committed and a blind retry can duplicate the write. Log the key and record for manual verification instead.

CRUD operations

The following example demonstrates create, update, read, and delete (CRUD) operations within a transaction. For the purposes of this example, the following records are held in a bin named sightings:

"sightings": [
    {"sighting":{"occurred":20200912,"reported":20200916,"posted":20201105,"report":{"city":"Kirkland","duration":"~30 minutes","shape":["circle"],"state":"WA","summary":"4 rotating orange lights in the Kingsgate area above the Safeway. Around 9pm the power went out in the Kingsgate area.  Four lights were spotted rotating above the local Safeway and surrounding streets.  They were rotating fast but staying relatively in the same spots.  Also described as orange lights. About thirty minutes later they disappeared.  The second they disappeared the power was restored.  Later a station of police from Woodinville and Kirkland came to guard the street where it happened.  They wouldn't let anyone go past the street, putting out search lights and flare signals so people couldn't drive past Safeway.  The police also would not let people walk past to go home."},"location":"\"{\"type\":\"Point\",\"coordinates\":[-122.1966441,47.69328259]}\""}}
    {"sighting":{"occurred":20200322,"reported":20200322,"posted":20200515,"report":{"city":"Pismo Beach","duration":"5 minutes","shape":["light"],"state":"CA","summary":"About 20 solid, bright lights moving at the same altitude, heading and speed.  Spaced perfectly apart flying over the ocean headed south."},"location":"\"{\"type\":\"Point\",\"coordinates\":[-120.6595,35.1546]}\""}}
    {"sighting":{"occurred":20200530,"reported":20200531,"posted":20200625,"report":{"city":"New York Staten Island","duration":"2 minutes","shape":["disk"],"state":"NY","summary":"Round shaped object observed over Staten Island NYC, while sitting in my back yard. My daughter also observed this object . Bright White shaped object moving fast from East to West . Observed over Graniteville, Staten Island towards the Elizabeth NJ area and appears to be fast. We then lost view of it due to the clouds."}}}]

Complete example:

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.aerospike.client.AerospikeClient;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.Bin;
import com.aerospike.client.Key;
import com.aerospike.client.Record;
import com.aerospike.client.ResultCode;
import com.aerospike.client.Txn;
import com.aerospike.client.Value;
import com.aerospike.client.cdt.CTX;
import com.aerospike.client.cdt.ListOperation;
import com.aerospike.client.cdt.ListReturnType;
import com.aerospike.client.cdt.MapOperation;
import com.aerospike.client.cdt.MapOrder;
import com.aerospike.client.cdt.MapPolicy;
import com.aerospike.client.cdt.MapWriteMode;
import com.aerospike.client.policy.RecordExistsAction;
import com.aerospike.client.policy.WritePolicy;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;

public class Transactions {
    public static void main(String ... args) {
        AerospikeClient client = new AerospikeClient("127.0.0.1", 3000);

        // Aerospike namespace, set, and key_id to be used for the Aerospike key
        String namespace = "sandbox";
        String set = "ufo";
        String binName = "sightings";
        int keyId = 6120, updateRecordIndex = 0;

        // Example JSON string
        String sightings =
                "[{\"sighting\":{\"occurred\":20200912,\"reported\":20200916,\"posted\":20201105,\"report\":{\"city\":\"Kirkland\",\"duration\":\"~30 minutes\",\"shape\":[\"circle\"],\"state\":\"WA\",\"summary\":\"4 rotating orange lights in the Kingsgate area above the Safeway. Around 9pm the power went out in the Kingsgate area.  Four lights were spotted rotating above the local Safeway and surrounding streets.  They were rotating fast but staying relatively in the same spots.  Also described as orange lights. About thirty minutes later they disappeared.  The second they disappeared the power was restored.  Later a station of police from Woodinville and Kirkland came to guard the street where it happened.  They wouldn't let anyone go past the street, putting out search lights and flare signals so people couldn't drive past Safeway.  The police also would not let people walk past to go home.\"},\"location\":\"\\\"{\\\"type\\\":\\\"Point\\\",\\\"coordinates\\\":[-122.1966441,47.69328259]}\\\"\"}},\n"+
                "{\"sighting\":{\"occurred\":20200530,\"reported\":20200531,\"posted\":20200625,\"report\":{\"city\":\"New York Staten Island\",\"duration\":\"2 minutes\",\"shape\":[\"disk\"],\"state\":\"NY\",\"summary\":\"Round shaped object observed over Staten Island NYC, while sitting in my back yard. My daughter also observed this object . Bright White shaped object moving fast from East to West . Observed over Graniteville, Staten Island towards the Elizabeth NJ area and appears to be fast. We then lost view of it due to the clouds.\"}}}]";

        // Convert string to Java Map
        List<Map<String, Object>> sightingMap = new Gson().fromJson(sightings, new TypeToken<ArrayList<HashMap<String, Object>>>(){}.getType());

        // Define Aerospike key
        Key key = new Key(namespace, set, keyId);

        // Set bins for "sightings" and "city"
        Bin sightingsBin = new Bin(binName, Value.get(sightingMap));

        // Start transaction
        Txn transaction = new Txn();

        // Create write policy
        WritePolicy writePolicy = client.copyWritePolicyDefault();
        writePolicy.recordExistsAction = RecordExistsAction.UPDATE;
        // Durable deletes are required for record removals inside a transaction.
        writePolicy.durableDelete = true;
        writePolicy.txn = transaction;

        System.out.printf("Begin transaction: %d%n", transaction.getId());

        // Result of the transaction body. Only promoted to operationResponse once
        // the commit is confirmed successful.
        Record response = null;
        Record operationResponse = null;

        try {
            response = performOperationWithTransaction(client, binName, updateRecordIndex, sightingMap, key, sightingsBin, writePolicy);
        } // Write failed
        catch (AerospikeException ae) {
            // Abort the transaction to roll back all changes and release locked records.
            // A new Txn is required for any subsequent attempt - this Txn can no longer be used once aborted.
            client.abort(transaction);
            response = null;

            if (ae.getResultCode() == ResultCode.MRT_BLOCKED) {
                // Transaction was blocked by another transaction - retry with a new Txn.
                System.out.printf("Write failed, transaction blocked %nError: %s%n", ae.getMessage());
            } else if (ae.getResultCode() == ResultCode.MRT_EXPIRED) {
                // Transaction expired before commit or abort - retry with a new Txn.
                System.out.printf("Write failed, transaction expired %nError: %s%n", ae.getMessage());
            } else {
                System.out.printf("Write failed, transaction aborted %nError: %s%n", ae.getMessage());
            }
        }
        catch (Throwable t) {
            // Handle non-Aerospike exception
            client.abort(transaction);
            response = null;
            System.out.printf("Write failed, transaction aborted %nError: %s%n", t.getMessage());
        }

        if (response != null) {
            try {
                // Commit transaction
                client.commit(transaction);
                operationResponse = response;
            }
            catch (AerospikeException.Commit ce) {
                if (ce.getInDoubt()) {
                    try {
                        // The commit result is uncertain - retry the commit itself
                        // (not the whole transaction) to resolve the outcome.
                        client.commit(transaction);
                        operationResponse = response;
                    }
                    catch (AerospikeException.Commit ce2) {
                        if (ce2.getInDoubt()) {
                            // Recommit still in doubt. The transaction outcome is unknown -
                            // log the key/record for manual verification and cleanup rather
                            // than assuming failure and retrying with a new Txn, which risks
                            // duplicating the write if the transaction actually committed.
                            System.out.printf(
                                "Recommit still in doubt. %nKey: %s%nRecord: %s%n",
                                key.userKey.toLong(),
                                new Gson().toJson(sightingMap)
                            );
                        }
                    }
                }
                else if (ce.getResultCode() == ResultCode.MRT_VERSION_MISMATCH) {
                    // The read-verify step failed because another command changed a record
                    // read by this transaction. commit() has already aborted and rolled back
                    // the transaction internally. Retry the entire transaction with a new Txn.
                    System.out.printf("Commit failed, transaction aborted %nError: %s%n", ce.getMessage());
                }
                else {
                    // Other commit failure. commit() has already aborted and rolled back the
                    // transaction internally. Retry the entire transaction with a new Txn.
                    System.out.printf("Commit failed, transaction aborted %nError: %s%n", ce.getMessage());
                }
            }
            catch (Throwable t) {
                // Handle non-Aerospike exception
                System.out.printf("Commit failed %nError: %s%n", t.getMessage());
            }
        }

        // Extract transaction results from records, if the transaction succeeded
        if (operationResponse != null) {
            List<?> results = (List<?>) operationResponse.getValue(binName); // Fetching operation results as list of individual operation result. In this case update, delete, and read

            Map<String, Object> removedRecord = (Map<String, Object>) results.get(1); // Fetching delete operation result. In our example we have asked for the removed record to be returned

            List<Map<String, Map<String, Map<String, Map<String, Object>>>>> binAfterOperation = (List<Map<String, Map<String, Map<String, Map<String, Object>>>>>) results.get(2); // Fetching read operation results. In our example this would be all the records in `sightings` bin

            System.out.printf(
                "Inserted Objects: %s%n",
                new Gson().toJson(sightingMap)
            );
            System.out.printf(
                "Updated `city` with value %s in object %s%n",
                binAfterOperation.get(0).get("sighting").get("report").get("city"), binAfterOperation.get(0)
            );
            System.out.printf(
                "Deleted from bin %s record %s%n",
                binName,
                new Gson().toJson(removedRecord)
            );
        } else {
            System.out.println("Transaction did not succeed");
        }

        // Close the client
        client.close();
    }

    private static Record performOperationWithTransaction(
        AerospikeClient client,
        String binName,
        int updateRecordIndex,
        List<Map<String, Object>> sightingMap,
        Key key,
        Bin sightingsBin,
        WritePolicy writePolicy) {

        // Writing record to Aerospike
        client.put(writePolicy, key, sightingsBin);
        System.out.printf("Created succeeded %nKey: %d%n", key.userKey.toLong());

        // Update record
        Record operationResponse = client.operate(writePolicy, key,
                MapOperation.put( // Record update first record in the list and assigning value `Seattle` to key `city`
                        new MapPolicy(MapOrder.KEY_VALUE_ORDERED, MapWriteMode.UPDATE),
                        binName, // Bin to be updated
                        Value.get("city"), // Key to be updated
                        Value.get("Seattle"), // Updated value
                        CTX.listIndex(updateRecordIndex), CTX.mapKey(Value.get("sighting")), CTX.mapKey(Value.get("report"))),
                ListOperation.removeByIndex(binName, 1, ListReturnType.VALUE), // Removing second record in list
                ListOperation.getRange(binName, 0, sightingMap.size()) // Returning remaining records
        );

        return operationResponse;
    }
}