* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
-
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import akka.actor.ActorSelection;
import akka.actor.Props;
-import akka.actor.UntypedActor;
+import akka.actor.UntypedAbstractActor;
import akka.dispatch.Dispatchers;
import akka.dispatch.Futures;
import akka.testkit.TestActorRef;
import com.codahale.metrics.Snapshot;
import com.codahale.metrics.Timer;
-import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.ArrayList;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
static class TestException extends RuntimeException {
}
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
@Mock
private Timer commitTimer;
public void setUp() {
MockitoAnnotations.initMocks(this);
- actorContext = new ActorContext(getSystem(), actorFactory.createActor(Props.create(DoNothingActor.class)),
+ actorUtils = new ActorUtils(getSystem(), actorFactory.createActor(Props.create(DoNothingActor.class)),
new MockClusterWrapper(), new MockConfiguration(), DatastoreContext.newBuilder().build(),
new PrimaryShardInfoFutureCache()) {
@Override
- public Timer getOperationTimer(String operationName) {
+ public Timer getOperationTimer(final String operationName) {
return commitTimer;
}
@Test
public void testCanCommitYesWithOneCohort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION)))), tx);
@Test
public void testCanCommitNoWithOneCohort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION)))), tx);
CanCommitTransactionReply.yes(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION))));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifyCohortActors();
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), false);
verifyCohortActors();
@Test(expected = TestException.class)
public void testCanCommitWithExceptionFailure() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(new TestException()))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
@Test(expected = IllegalArgumentException.class)
public void testCanCommitWithInvalidResponseType() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit("invalid"))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
newCohortInfo(new CohortActor.Builder(tx)),
newCohortInfoWithFailedFuture(new TestException()),
newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
propagateExecutionExceptionCause(proxy.canCommit());
}
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(new TestException())));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
@Test(expected = IllegalArgumentException.class)
public void testCommitWithInvalidResponseType() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext,
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils,
Arrays.asList(newCohortInfo(new CohortActor.Builder(tx)
.expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).expectCommit("invalid"))), tx);
@Test
public void testAbort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectAbort(
AbortTransactionReply.instance(CURRENT_VERSION)))), tx);
@Test
public void testAbortWithFailure() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectAbort(new RuntimeException("mock")))), tx);
// The exception should not get propagated.
public void testAbortWithFailedCohortFuture() throws Exception {
List<CohortInfo> cohorts = Arrays.asList(
newCohortInfoWithFailedFuture(new TestException()), newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifySuccessfulFuture(proxy.abort());
verifyCohortActors();
@Test
public void testWithNoCohorts() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext,
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils,
Collections.<CohortInfo>emptyList(), tx);
verifyCanCommit(proxy.canCommit(), true);
verifyCohortActors();
}
- private void propagateExecutionExceptionCause(ListenableFuture<?> future) throws Exception {
-
+ @SuppressWarnings("checkstyle:avoidHidingCauseException")
+ private void propagateExecutionExceptionCause(final ListenableFuture<?> future) throws Exception {
try {
future.get(5, TimeUnit.SECONDS);
fail("Expected ExecutionException");
} catch (ExecutionException e) {
verifyCohortActors();
- Throwables.propagateIfInstanceOf(e.getCause(), Exception.class);
- Throwables.propagate(e.getCause());
+ Throwables.propagateIfPossible(e.getCause(), Exception.class);
+ throw new RuntimeException(e.getCause());
}
}
- private CohortInfo newCohortInfo(CohortActor.Builder builder, final short version) {
+ private CohortInfo newCohortInfo(final CohortActor.Builder builder, final short version) {
TestActorRef<CohortActor> actor = actorFactory.createTestActor(builder.props()
.withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("cohort"));
cohortActors.add(actor);
return new CohortInfo(Futures.successful(getSystem().actorSelection(actor.path())), () -> version);
}
- private CohortInfo newCohortInfo(CohortActor.Builder builder) {
+ private CohortInfo newCohortInfo(final CohortActor.Builder builder) {
return newCohortInfo(builder, CURRENT_VERSION);
}
- private static CohortInfo newCohortInfoWithFailedFuture(Exception failure) {
+ private static CohortInfo newCohortInfoWithFailedFuture(final Exception failure) {
return new CohortInfo(Futures.<ActorSelection>failed(failure), () -> CURRENT_VERSION);
}
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private <T> T verifySuccessfulFuture(ListenableFuture<T> future) throws Exception {
+ private <T> T verifySuccessfulFuture(final ListenableFuture<T> future) throws Exception {
try {
return future.get(5, TimeUnit.SECONDS);
} catch (Exception e) {
}
}
- private void verifyCanCommit(ListenableFuture<Boolean> future, boolean expected) throws Exception {
+ private void verifyCanCommit(final ListenableFuture<Boolean> future, final boolean expected) throws Exception {
Boolean actual = verifySuccessfulFuture(future);
assertEquals("canCommit", expected, actual);
}
- private static class CohortActor extends UntypedActor {
+ private static class CohortActor extends UntypedAbstractActor {
private final Builder builder;
private final AtomicInteger canCommitCount = new AtomicInteger();
private final AtomicInteger commitCount = new AtomicInteger();
private final AtomicInteger abortCount = new AtomicInteger();
private volatile AssertionError assertionError;
- private CohortActor(Builder builder) {
+ CohortActor(final Builder builder) {
this.builder = builder;
}
@Override
- public void onReceive(Object message) {
+ public void onReceive(final Object message) {
if (CanCommitTransaction.isSerializedType(message)) {
canCommitCount.incrementAndGet();
onMessage("CanCommitTransaction", message, CanCommitTransaction.fromSerializable(message),
}
}
- private void onMessage(String name, Object rawMessage, AbstractThreePhaseCommitMessage actualMessage,
- Class<?> expType, Object reply) {
+ private void onMessage(final String name, final Object rawMessage,
+ final AbstractThreePhaseCommitMessage actualMessage, final Class<?> expType, final Object reply) {
try {
assertNotNull("Unexpected " + name, expType);
assertEquals(name + " type", expType, rawMessage.getClass());
- assertEquals(name + " transactionId", builder.transactionId, actualMessage.getTransactionID());
+ assertEquals(name + " transactionId", builder.transactionId, actualMessage.getTransactionId());
if (reply instanceof Throwable) {
getSender().tell(new akka.actor.Status.Failure((Throwable)reply), self());
private Object abortReply;
private final TransactionIdentifier transactionId;
- Builder(TransactionIdentifier transactionId) {
- this.transactionId = Preconditions.checkNotNull(transactionId);
+ Builder(final TransactionIdentifier transactionId) {
+ this.transactionId = requireNonNull(transactionId);
}
- Builder expectCanCommit(Class<?> newExpCanCommitType, Object newCanCommitReply) {
+ Builder expectCanCommit(final Class<?> newExpCanCommitType, final Object newCanCommitReply) {
this.expCanCommitType = newExpCanCommitType;
this.canCommitReply = newCanCommitReply;
return this;
}
- Builder expectCanCommit(Object newCanCommitReply) {
+ Builder expectCanCommit(final Object newCanCommitReply) {
return expectCanCommit(CanCommitTransaction.class, newCanCommitReply);
}
- Builder expectCommit(Class<?> newExpCommitType, Object newCommitReply) {
+ Builder expectCommit(final Class<?> newExpCommitType, final Object newCommitReply) {
this.expCommitType = newExpCommitType;
this.commitReply = newCommitReply;
return this;
}
- Builder expectCommit(Object newCommitReply) {
+ Builder expectCommit(final Object newCommitReply) {
return expectCommit(CommitTransaction.class, newCommitReply);
}
- Builder expectAbort(Class<?> newExpAbortType, Object newAbortReply) {
+ Builder expectAbort(final Class<?> newExpAbortType, final Object newAbortReply) {
this.expAbortType = newExpAbortType;
this.abortReply = newAbortReply;
return this;
}
- Builder expectAbort(Object newAbortReply) {
+ Builder expectAbort(final Object newAbortReply) {
return expectAbort(AbortTransaction.class, newAbortReply);
}