ready.getTransactionID(), ready.getTxnClientVersion());
ShardDataTreeCohort cohort = ready.getTransaction().ready();
- CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort);
+ CohortEntry cohortEntry = new CohortEntry(ready.getTransactionID(), cohort, ready.getTxnClientVersion());
cohortCache.put(ready.getTransactionID(), cohortEntry);
if(!queueCohortEntry(cohortEntry, sender, shard)) {
if(cohortEntry == null) {
cohortEntry = new CohortEntry(batched.getTransactionID(),
dataTree.newReadWriteTransaction(batched.getTransactionID(),
- batched.getTransactionChainID()));
+ batched.getTransactionChainID()), batched.getVersion());
cohortCache.put(batched.getTransactionID(), cohortEntry);
}
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);
+ final CohortEntry cohortEntry = new CohortEntry(message.getTransactionID(), cohort,
+ DataStoreVersions.CURRENT_VERSION);
cohortCache.put(message.getTransactionID(), cohortEntry);
cohortEntry.setDoImmediateCommit(message.isDoCommitOnReady());
"Can Commit failed, no detailed cause available.")), cohortEntry.getShard().self());
}
} else {
+ // FIXME - use caller's version
cohortEntry.getReplySender().tell(
- canCommit ? CanCommitTransactionReply.YES.toSerializable() :
- CanCommitTransactionReply.NO.toSerializable(), cohortEntry.getShard().self());
+ canCommit ? CanCommitTransactionReply.yes(cohortEntry.getClientVersion()).toSerializable() :
+ CanCommitTransactionReply.no(cohortEntry.getClientVersion()).toSerializable(),
+ cohortEntry.getShard().self());
}
} catch (Exception e) {
log.debug("{}: An exception occurred during canCommit", name, e);
shard.getShardMBean().incrementAbortTransactionsCount();
if(sender != null) {
- sender.tell(new AbortTransactionReply().toSerializable(), self);
+ sender.tell(AbortTransactionReply.instance(cohortEntry.getClientVersion()).toSerializable(), self);
}
} catch (Exception e) {
log.error("{}: An exception happened during abort", name, e);
private final Stopwatch lastAccessTimer = Stopwatch.createStarted();
private int totalBatchedModificationsReceived;
private boolean aborted;
+ private final short clientVersion;
- CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction) {
+ CohortEntry(String transactionID, ReadWriteShardDataTreeTransaction transaction, short clientVersion) {
this.transaction = Preconditions.checkNotNull(transaction);
this.transactionID = transactionID;
+ this.clientVersion = clientVersion;
}
- CohortEntry(String transactionID, ShardDataTreeCohort cohort) {
+ CohortEntry(String transactionID, ShardDataTreeCohort cohort, short clientVersion) {
this.transactionID = transactionID;
this.cohort = cohort;
this.transaction = null;
+ this.clientVersion = clientVersion;
}
void updateLastAccessTime() {
return transactionID;
}
+ short getClientVersion() {
+ return clientVersion;
+ }
+
DataTreeCandidate getCandidate() {
return cohort.getCandidate();
}