+ log.debug("{}: Applying {} batched modifications for Tx {}", name,
+ batched.getModifications().size(), 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;
+ }
+
+ if(log.isDebugEnabled()) {
+ log.debug("{}: Readying Tx {}, client version {}", name,
+ batched.getTransactionID(), batched.getVersion());
+ }
+
+ cohortEntry.ready(cohortDecorator, batched.isDoCommitOnReady());
+
+ if(batched.isDoCommitOnReady()) {
+ cohortEntry.setReplySender(sender);
+ cohortEntry.setShard(shard);
+ handleCanCommit(cohortEntry);
+ } else {
+ sender.tell(readyTransactionReply(shard), shard.self());
+ }
+ } else {
+ sender.tell(new BatchedModificationsReply(batched.getModifications().size()), shard.self());
+ }
+ }
+
+ /**
+ * 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 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(),
+ message.getTransactionID());
+ final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort);
+ cohortCache.put(message.getTransactionID(), cohortEntry);
+ cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
+
+ if(!queueCohortEntry(cohortEntry, sender, shard)) {
+ return;