*
* @author Thomas Pantelis
*/
-public class ShardCommitCoordinator {
+class ShardCommitCoordinator {
// Interface hook for unit tests to replace or decorate the DOMStoreThreePhaseCommitCohorts.
public interface CohortDecorator {
private ReadyTransactionReply readyTransactionReply;
- public ShardCommitCoordinator(ShardDataTree dataTree,
+ ShardCommitCoordinator(ShardDataTree dataTree,
long cacheExpiryTimeoutInMillis, int queueCapacity, ActorRef shardActor, Logger log, String name) {
this.queueCapacity = queueCapacity;
this.cacheExpiryTimeoutInMillis = cacheExpiryTimeoutInMillis;
}
- public void setQueueCapacity(int queueCapacity) {
+ void setQueueCapacity(int queueCapacity) {
this.queueCapacity = queueCapacity;
}
/**
* This method is called to ready a transaction that was prepared by ShardTransaction actor. It caches
* the prepared cohort entry for the given transactions ID in preparation for the subsequent 3-phase commit.
+ *
+ * @param ready the ForwardedReadyTransaction message to process
+ * @param sender the sender of the message
+ * @param shard the transaction's shard actor
*/
- public void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
+ void handleForwardedReadyTransaction(ForwardedReadyTransaction ready, ActorRef sender, Shard shard) {
log.debug("{}: Readying transaction {}, client version {}", name,
ready.getTransactionID(), ready.getTxnClientVersion());
* DOMStoreWriteTransaction, one is created. The batched modifications are applied to the write Tx. If
* the BatchedModifications is ready to commit then a DOMStoreThreePhaseCommitCohort is created.
*
- * @param batched the BatchedModifications
- * @param shardActor the transaction's shard actor
- *
- * @throws ExecutionException if an error occurs loading the cache
+ * @param batched the BatchedModifications message to process
+ * @param sender the sender of the message
+ * @param shard the transaction's shard actor
*/
- void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard)
- throws ExecutionException {
+ void handleBatchedModifications(BatchedModifications batched, ActorRef sender, Shard shard) {
CohortEntry cohortEntry = cohortCache.get(batched.getTransactionID());
if(cohortEntry == null) {
cohortEntry = new CohortEntry(batched.getTransactionID(),
cohortEntry.applyModifications(batched.getModifications());
if(batched.isReady()) {
+ if(cohortEntry.getLastBatchedModificationsException() != null) {
+ cohortCache.remove(cohortEntry.getTransactionID());
+ throw cohortEntry.getLastBatchedModificationsException();
+ }
+
+ if(cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) {
+ cohortCache.remove(cohortEntry.getTransactionID());
+ throw new IllegalStateException(String.format(
+ "The total number of batched messages received %d does not match the number sent %d",
+ cohortEntry.getTotalBatchedModificationsReceived(), batched.getTotalMessagesSent()));
+ }
+
if(!queueCohortEntry(cohortEntry, sender, shard)) {
return;
}
* This method handles {@link ReadyLocalTransaction} message. All transaction modifications have
* been prepared beforehand by the sender and we just need to drive them through into the dataTree.
*
- * @param message
- * @param sender
- * @param shard
+ * @param message the ReadyLocalTransaction message to process
+ * @param sender the sender of the message
+ * @param shard the transaction's shard actor
*/
void handleReadyLocalTransaction(ReadyLocalTransaction message, ActorRef sender, Shard shard) {
- final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification());
+ final ShardDataTreeCohort cohort = new SimpleShardDataTreeCohort(dataTree, message.getModification(),
+ message.getTransactionID());
final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort);
cohortCache.put(message.getTransactionID(), cohortEntry);
cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
/**
* This method handles the canCommit phase for a transaction.
*
- * @param canCommit the CanCommitTransaction message
- * @param sender the actor that sent the message
+ * @param transactionID the ID of the transaction to canCommit
+ * @param sender the actor to which to send the response
* @param shard the transaction's shard actor
*/
- public void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
+ void handleCanCommit(String transactionID, final ActorRef sender, final Shard shard) {
// Lookup the cohort entry that was cached previously (or should have been) by
// transactionReady (via the ForwardedReadyTransaction message).
final CohortEntry cohortEntry = cohortCache.get(transactionID);
return success;
}
+ /**
+ * 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.
private final String transactionID;
private ShardDataTreeCohort cohort;
private final ReadWriteShardDataTreeTransaction transaction;
+ private RuntimeException lastBatchedModificationsException;
private ActorRef replySender;
private Shard shard;
private boolean doImmediateCommit;
private final Stopwatch lastAccessTimer = Stopwatch.createStarted();
+ private int totalBatchedModificationsReceived;
CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction) {
this.transaction = Preconditions.checkNotNull(transaction);
return cohort;
}
+ int getTotalBatchedModificationsReceived() {
+ return totalBatchedModificationsReceived;
+ }
+
+ RuntimeException getLastBatchedModificationsException() {
+ return lastBatchedModificationsException;
+ }
+
void applyModifications(Iterable<Modification> modifications) {
- for (Modification modification : modifications) {
- modification.apply(transaction.getSnapshot());
+ totalBatchedModificationsReceived++;
+ if(lastBatchedModificationsException == null) {
+ for (Modification modification : modifications) {
+ try {
+ modification.apply(transaction.getSnapshot());
+ } catch (RuntimeException e) {
+ lastBatchedModificationsException = e;
+ throw e;
+ }
+ }
}
}
Code Review / controller.git / blobdiff