import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import javax.annotation.Nonnull;
log.debug("{}: Readying transaction {}, client version {}", name,
ready.getTransactionId(), ready.getTxnClientVersion());
- final ShardDataTreeCohort cohort = ready.getTransaction().ready();
+ final ShardDataTreeCohort cohort = ready.getTransaction().ready(ready.getParticipatingShardNames());
final CohortEntry cohortEntry = CohortEntry.createReady(cohort, ready.getTxnClientVersion());
cohortCache.put(cohortEntry.getTransactionId(), cohortEntry);
}
cohortEntry.setDoImmediateCommit(batched.isDoCommitOnReady());
- cohortEntry.ready(cohortDecorator);
+ cohortEntry.ready(batched.getParticipatingShardNames(), cohortDecorator);
if (batched.isDoCommitOnReady()) {
cohortEntry.setReplySender(sender);
* @param shard the transaction's shard actor
*/
void handleReadyLocalTransaction(final ReadyLocalTransaction message, final ActorRef sender, final Shard shard) {
- final ShardDataTreeCohort cohort = dataTree.createReadyCohort(message.getTransactionId(),
- message.getModification());
+ final TransactionIdentifier txId = message.getTransactionId();
+ final ShardDataTreeCohort cohort = dataTree.newReadyCohort(txId, message.getModification(),
+ message.getParticipatingShardNames());
final CohortEntry cohortEntry = CohortEntry.createReady(cohort, DataStoreVersions.CURRENT_VERSION);
cohortCache.put(cohortEntry.getTransactionId(), cohortEntry);
cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
- log.debug("{}: Applying local modifications for Tx {}", name, message.getTransactionId());
+ log.debug("{}: Applying local modifications for Tx {}", name, txId);
if (message.isDoCommitOnReady()) {
cohortEntry.setReplySender(sender);
BatchedModifications last = newModifications.getLast();
last.setDoCommitOnReady(from.isDoCommitOnReady());
- last.setReady(from.isReady());
+ if (from.isReady()) {
+ last.setReady(from.getParticipatingShardNames());
+ }
last.setTotalMessagesSent(newModifications.size());
return newModifications;
}
final TransactionIdentifier txId = cohortEntry.getTransactionId();
log.debug("{}: Transaction {} committed as {}, sending response to {}", persistenceId(), txId, result,
sender);
+ cohortEntry.getShard().getDataStore().purgeTransaction(txId, null);
cohortCache.remove(cohortEntry.getTransactionId());
sender.tell(CommitTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(),
@Override
public void onFailure(final Throwable failure) {
- log.error("{}, An exception occurred while committing transaction {}", persistenceId(),
- cohortEntry.getTransactionId(), failure);
+ final TransactionIdentifier txId = cohortEntry.getTransactionId();
+ log.error("{}, An exception occurred while committing transaction {}", persistenceId(), txId, failure);
+ cohortEntry.getShard().getDataStore().purgeTransaction(txId, null);
cohortCache.remove(cohortEntry.getTransactionId());
sender.tell(new Failure(failure), cohortEntry.getShard().self());
cohortEntry.abort(new FutureCallback<Void>() {
@Override
public void onSuccess(final Void result) {
+ shard.getDataStore().purgeTransaction(cohortEntry.getTransactionId(), null);
+
if (sender != null) {
sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
}
@Override
public void onFailure(final Throwable failure) {
log.error("{}: An exception happened during abort", name, failure);
+ shard.getDataStore().purgeTransaction(cohortEntry.getTransactionId(), null);
if (sender != null) {
sender.tell(new Failure(failure), self);
}
void checkForExpiredTransactions(final long timeout, final Shard shard) {
- Iterator<CohortEntry> iter = cohortCache.values().iterator();
- while (iter.hasNext()) {
- CohortEntry cohortEntry = iter.next();
- if (cohortEntry.isFailed()) {
- iter.remove();
- }
- }
+ cohortCache.values().removeIf(CohortEntry::isFailed);
}
void abortPendingTransactions(final String reason, final Shard shard) {
if (last != null) {
final boolean immediate = cohortEntry.isDoImmediateCommit();
last.setDoCommitOnReady(immediate);
- last.setReady(true);
+ last.setReady(cohortEntry.getParticipatingShardNames());
last.setTotalMessagesSent(newMessages.size());
messages.addAll(newMessages);