import akka.dispatch.OnComplete;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
-import javax.annotation.Nonnull;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
private final DataTreeModification modification;
private final ActorContext actorContext;
private final ActorSelection leader;
- private Exception operationError;
+ private final Exception operationError;
protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction, final DataTreeModification modification) {
this.leader = Preconditions.checkNotNull(leader);
this.transaction = Preconditions.checkNotNull(transaction);
this.modification = Preconditions.checkNotNull(modification);
+ this.operationError = null;
}
protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
return actorContext.executeOperationAsync(leader, message, actorContext.getTransactionCommitOperationTimeout());
}
- void setOperationError(@Nonnull Exception operationError) {
- if (this.operationError != null) {
- LOG.info("Cohort {} already had operation error", this, this.operationError);
- }
-
- this.operationError = Preconditions.checkNotNull(operationError);
- }
-
Future<ActorSelection> initiateCoordinatedCommit() {
final Future<Object> messageFuture = initiateCommit(false);
final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorContext,
import akka.actor.ActorSelection;
import com.google.common.base.Preconditions;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.sal.core.spi.data.AbstractSnapshotBackedTransactionChain;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
@SuppressWarnings("unchecked")
@Override
- public LocalThreePhaseCommitCohort onTransactionReady(DOMStoreWriteTransaction tx) {
+ public LocalThreePhaseCommitCohort onTransactionReady(@Nonnull DOMStoreWriteTransaction tx,
+ @Nullable Exception operationError) {
+ if(operationError != null) {
+ return new LocalChainThreePhaseCommitCohort((SnapshotBackedWriteTransaction<TransactionIdentifier>)tx,
+ operationError);
+ }
+
try {
return (LocalThreePhaseCommitCohort) tx.ready();
} catch (Exception e) {
private LocalThreePhaseCommitCohort ready() {
logModificationCount();
- LocalThreePhaseCommitCohort cohort = readySupport.onTransactionReady(getWriteDelegate());
- if (operationError != null) {
- cohort.setOperationError(operationError);
- }
- return cohort;
+ return readySupport.onTransactionReady(getWriteDelegate(), operationError);
}
@Override
import akka.actor.ActorSelection;
import com.google.common.base.Preconditions;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
@SuppressWarnings("unchecked")
@Override
- public LocalThreePhaseCommitCohort onTransactionReady(DOMStoreWriteTransaction tx) {
+ public LocalThreePhaseCommitCohort onTransactionReady(@Nonnull DOMStoreWriteTransaction tx,
+ @Nullable Exception operationError) {
+ if(operationError != null) {
+ return new LocalThreePhaseCommitCohort(actorContext, leader,
+ (SnapshotBackedWriteTransaction<TransactionIdentifier>)tx, operationError);
+ }
+
try {
return (LocalThreePhaseCommitCohort) tx.ready();
} catch (Exception e) {
*/
package org.opendaylight.controller.cluster.datastore;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
/**
* @author Thomas Pantelis
*/
interface LocalTransactionReadySupport {
- LocalThreePhaseCommitCohort onTransactionReady(DOMStoreWriteTransaction tx);
+ LocalThreePhaseCommitCohort onTransactionReady(@Nonnull DOMStoreWriteTransaction tx,
+ @Nullable Exception operationError);
}
public void testReady() {
final LocalThreePhaseCommitCohort mockCohort = mock(LocalThreePhaseCommitCohort.class);
doReturn(akka.dispatch.Futures.successful(null)).when(mockCohort).initiateCoordinatedCommit();
- doReturn(mockCohort).when(mockReadySupport).onTransactionReady(readWriteTransaction);
+ doReturn(mockCohort).when(mockReadySupport).onTransactionReady(readWriteTransaction, null);
Future<ActorSelection> future = localTransactionContext.readyTransaction();
assertTrue(future.isCompleted());
- verify(mockReadySupport).onTransactionReady(readWriteTransaction);
+ verify(mockReadySupport).onTransactionReady(readWriteTransaction, null);
}
@Test
private void doReadyWithExpectedError(RuntimeException expError) {
LocalThreePhaseCommitCohort mockCohort = mock(LocalThreePhaseCommitCohort.class);
doReturn(akka.dispatch.Futures.successful(null)).when(mockCohort).initiateCoordinatedCommit();
- doReturn(mockCohort).when(mockReadySupport).onTransactionReady(readWriteTransaction);
+ doReturn(mockCohort).when(mockReadySupport).onTransactionReady(readWriteTransaction, expError);
localTransactionContext.readyTransaction();
-
- verify(mockCohort).setOperationError(expError);
}
}