*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.actor.Status.Failure;
import akka.serialization.Serialization;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.primitives.UnsignedLong;
import com.google.common.util.concurrent.FutureCallback;
import java.util.ArrayDeque;
ShardCommitCoordinator(final ShardDataTree dataTree, final Logger log, final String name) {
this.log = log;
this.name = name;
- this.dataTree = Preconditions.checkNotNull(dataTree);
+ this.dataTree = requireNonNull(dataTree);
}
int getCohortCacheSize() {
*/
void handleBatchedModifications(final BatchedModifications batched, final ActorRef sender, final Shard shard) {
CohortEntry cohortEntry = cohortCache.get(batched.getTransactionId());
- if (cohortEntry == null) {
+ if (cohortEntry == null || cohortEntry.isSealed()) {
cohortEntry = CohortEntry.createOpen(dataTree.newReadWriteTransaction(batched.getTransactionId()),
batched.getVersion());
cohortCache.put(cohortEntry.getTransactionId(), cohortEntry);
}
if (log.isDebugEnabled()) {
- log.debug("{}: Readying Tx {}, client version {}", name,
- batched.getTransactionId(), batched.getVersion());
+ log.debug("{}: Readying Tx {} of {} operations, client version {}", name,
+ batched.getTransactionId(), cohortEntry.getTotalOperationsProcessed(), batched.getVersion());
}
cohortEntry.setDoImmediateCommit(batched.isDoCommitOnReady());
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(),
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(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);