+ /**
+ * This method handles the preCommit and commit phases for a transaction.
+ *
+ * @param transactionID the ID of the transaction to commit
+ * @param sender the actor to which to send the response
+ * @param shard the transaction's shard actor
+ * @return true if the transaction was successfully prepared, false otherwise.
+ */
+ boolean handleCommit(final String transactionID, final ActorRef sender, final Shard shard) {
+ // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
+ // this transaction.
+ final CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
+ if(cohortEntry == null) {
+ // We're not the current Tx - the Tx was likely expired b/c it took too long in
+ // between the canCommit and commit messages.
+ IllegalStateException ex = new IllegalStateException(
+ String.format("%s: Cannot commit transaction %s - it is not the current transaction",
+ name, transactionID));
+ log.error(ex.getMessage());
+ sender.tell(new akka.actor.Status.Failure(ex), shard.self());
+ return false;
+ }
+
+ cohortEntry.setReplySender(sender);
+ return doCommit(cohortEntry);
+ }
+
+ void handleAbort(final String transactionID, final ActorRef sender, final Shard shard) {
+ CohortEntry cohortEntry = getCohortEntryIfCurrent(transactionID);
+ if(cohortEntry != null) {
+ // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
+ // aborted during replication in which case we may still commit locally if replication
+ // succeeds.
+ currentTransactionComplete(transactionID, false);
+ } else {
+ cohortEntry = getAndRemoveCohortEntry(transactionID);
+ }
+
+ if(cohortEntry == null) {
+ return;
+ }
+
+ log.debug("{}: Aborting transaction {}", name, transactionID);