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;
* @param ready the ForwardedReadyTransaction message to process
* @param sender the sender of the message
* @param shard the transaction's shard actor
- * @param schema
*/
void handleForwardedReadyTransaction(final ForwardedReadyTransaction ready, final ActorRef sender,
final Shard shard) {
log.debug("{}: Readying transaction {}, client version {}", name,
- ready.getTransactionID(), ready.getTxnClientVersion());
+ 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);
+ cohortCache.put(cohortEntry.getTransactionId(), cohortEntry);
if (ready.isDoImmediateCommit()) {
cohortEntry.setDoImmediateCommit(true);
* @param sender the sender of the message
*/
void handleBatchedModifications(final BatchedModifications batched, final ActorRef sender, final Shard shard) {
- CohortEntry cohortEntry = cohortCache.get(batched.getTransactionID());
+ CohortEntry cohortEntry = cohortCache.get(batched.getTransactionId());
if (cohortEntry == null) {
- cohortEntry = CohortEntry.createOpen(dataTree.newReadWriteTransaction(batched.getTransactionID()),
+ cohortEntry = CohortEntry.createOpen(dataTree.newReadWriteTransaction(batched.getTransactionId()),
batched.getVersion());
- cohortCache.put(cohortEntry.getTransactionID(), cohortEntry);
+ cohortCache.put(cohortEntry.getTransactionId(), cohortEntry);
}
if (log.isDebugEnabled()) {
log.debug("{}: Applying {} batched modifications for Tx {}", name,
- batched.getModifications().size(), batched.getTransactionID());
+ batched.getModifications().size(), batched.getTransactionId());
}
cohortEntry.applyModifications(batched.getModifications());
if (batched.isReady()) {
if (cohortEntry.getLastBatchedModificationsException() != null) {
- cohortCache.remove(cohortEntry.getTransactionID());
+ cohortCache.remove(cohortEntry.getTransactionId());
throw cohortEntry.getLastBatchedModificationsException();
}
if (cohortEntry.getTotalBatchedModificationsReceived() != batched.getTotalMessagesSent()) {
- cohortCache.remove(cohortEntry.getTransactionID());
+ 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 (log.isDebugEnabled()) {
log.debug("{}: Readying Tx {}, client version {}", name,
- batched.getTransactionID(), batched.getVersion());
+ batched.getTransactionId(), batched.getVersion());
}
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);
+ 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);
Collection<BatchedModifications> createForwardedBatchedModifications(final BatchedModifications from,
final int maxModificationsPerBatch) {
- CohortEntry cohortEntry = cohortCache.remove(from.getTransactionID());
+ CohortEntry cohortEntry = cohortCache.remove(from.getTransactionId());
if (cohortEntry == null || cohortEntry.getTransaction() == null) {
return Collections.singletonList(from);
}
cohortEntry.getTransaction().getSnapshot().applyToCursor(new AbstractBatchedModificationsCursor() {
@Override
protected BatchedModifications getModifications() {
- if (newModifications.isEmpty() ||
- newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
- newModifications.add(new BatchedModifications(from.getTransactionID(), from.getVersion()));
+ if (newModifications.isEmpty()
+ || newModifications.getLast().getModifications().size() >= maxModificationsPerBatch) {
+ newModifications.add(new BatchedModifications(from.getTransactionId(), from.getVersion()));
}
return newModifications.getLast();
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;
}
cohortEntry.canCommit(new FutureCallback<Void>() {
@Override
public void onSuccess(final Void result) {
- log.debug("{}: canCommit for {}: success", name, cohortEntry.getTransactionID());
+ log.debug("{}: canCommit for {}: success", name, cohortEntry.getTransactionId());
if (cohortEntry.isDoImmediateCommit()) {
doCommit(cohortEntry);
}
@Override
- public void onFailure(final Throwable t) {
+ public void onFailure(final Throwable failure) {
log.debug("{}: An exception occurred during canCommit for {}: {}", name,
- cohortEntry.getTransactionID(), t);
+ cohortEntry.getTransactionId(), failure);
- cohortCache.remove(cohortEntry.getTransactionID());
- cohortEntry.getReplySender().tell(new Failure(t), cohortEntry.getShard().self());
+ cohortCache.remove(cohortEntry.getTransactionId());
+ cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
}
});
}
handleCanCommit(cohortEntry);
}
- private void doCommit(final CohortEntry cohortEntry) {
- log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionID());
+ void doCommit(final CohortEntry cohortEntry) {
+ log.debug("{}: Committing transaction {}", name, cohortEntry.getTransactionId());
// We perform the preCommit phase here atomically with the commit phase. This is an
// optimization to eliminate the overhead of an extra preCommit message. We lose front-end
}
@Override
- public void onFailure(final Throwable t) {
+ public void onFailure(final Throwable failure) {
log.error("{} An exception occurred while preCommitting transaction {}", name,
- cohortEntry.getTransactionID(), t);
+ cohortEntry.getTransactionId(), failure);
- cohortCache.remove(cohortEntry.getTransactionID());
- cohortEntry.getReplySender().tell(new Failure(t), cohortEntry.getShard().self());
+ cohortCache.remove(cohortEntry.getTransactionId());
+ cohortEntry.getReplySender().tell(new Failure(failure), cohortEntry.getShard().self());
}
});
}
- private void finishCommit(@Nonnull final ActorRef sender, @Nonnull final CohortEntry cohortEntry) {
- log.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionID());
+ void finishCommit(@Nonnull final ActorRef sender, @Nonnull final CohortEntry cohortEntry) {
+ log.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionId());
cohortEntry.commit(new FutureCallback<UnsignedLong>() {
@Override
public void onSuccess(final UnsignedLong result) {
- final TransactionIdentifier txId = cohortEntry.getTransactionID();
+ 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());
+ cohortCache.remove(cohortEntry.getTransactionId());
sender.tell(CommitTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(),
cohortEntry.getShard().self());
}
@Override
- public void onFailure(final Throwable t) {
- log.error("{}, An exception occurred while committing transaction {}", persistenceId(),
- cohortEntry.getTransactionID(), t);
+ public void onFailure(final Throwable 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(t), cohortEntry.getShard().self());
+ cohortCache.remove(cohortEntry.getTransactionId());
+ sender.tell(new Failure(failure), cohortEntry.getShard().self());
}
});
}
doCommit(cohortEntry);
}
+ @SuppressWarnings("checkstyle:IllegalCatch")
void handleAbort(final Identifier transactionID, final ActorRef sender, final Shard shard) {
CohortEntry cohortEntry = cohortCache.remove(transactionID);
if (cohortEntry == null) {
log.debug("{}: Aborting transaction {}", name, transactionID);
final ActorRef self = shard.getSelf();
- try {
- cohortEntry.abort();
-
- shard.getShardMBean().incrementAbortTransactionsCount();
+ 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);
+ if (sender != null) {
+ sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
+ }
}
- } catch (Exception e) {
- log.error("{}: An exception happened during abort", name, e);
- if (sender != null) {
- sender.tell(new Failure(e), 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);
+ }
}
- }
+ });
+
+ shard.getShardMBean().incrementAbortTransactionsCount();
}
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) {
}
// Allocate a new message
- final BatchedModifications ret = new BatchedModifications(cohortEntry.getTransactionID(),
+ final BatchedModifications ret = new BatchedModifications(cohortEntry.getTransactionId(),
cohortEntry.getClientVersion());
newMessages.add(ret);
return ret;
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);
switch (cohort.getState()) {
case CAN_COMMIT_COMPLETE:
case CAN_COMMIT_PENDING:
- messages.add(new CanCommitTransaction(cohortEntry.getTransactionID(),
+ messages.add(new CanCommitTransaction(cohortEntry.getTransactionId(),
cohortEntry.getClientVersion()));
break;
case PRE_COMMIT_COMPLETE:
case PRE_COMMIT_PENDING:
- messages.add(new CommitTransaction(cohortEntry.getTransactionID(),
+ messages.add(new CommitTransaction(cohortEntry.getTransactionId(),
cohortEntry.getClientVersion()));
break;
default:
Code Review / controller.git / blobdiff