</dependency>
<!-- Test Dependencies -->
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
this.getContext(), id, new ElectionTermImpl(persistentProvider, id, LOG),
-1, -1, peerAddresses,
configParams.isPresent() ? configParams.get() : new DefaultConfigParamsImpl(),
- delegatingPersistenceProvider, this::handleApplyState, LOG);
+ delegatingPersistenceProvider, this::handleApplyState, LOG, this::executeInSelf);
context.setPayloadVersion(payloadVersion);
context.setReplicatedLog(ReplicatedLogImpl.newInstance(context));
import com.google.common.annotations.VisibleForTesting;
import java.util.Collection;
import java.util.Optional;
+import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.LongSupplier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.io.FileBackedOutputStreamFactory;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
*/
ActorRef getActor();
+ /**
+ * Return an Executor which is guaranteed to run tasks in the context of {@link #getActor()}.
+ *
+ * @return An executor.
+ */
+ @NonNull Executor getExecutor();
+
/**
* The akka Cluster singleton for the actor system if one is configured.
*
* 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.raft;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorContext;
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Props;
import akka.cluster.Cluster;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.Executor;
import java.util.function.Consumer;
import java.util.function.LongSupplier;
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.io.FileBackedOutputStreamFactory;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
private final ActorContext context;
+ private final @NonNull Executor executor;
+
private final String id;
private final ElectionTerm termInformation;
private RaftActorLeadershipTransferCohort leadershipTransferCohort;
public RaftActorContextImpl(final ActorRef actor, final ActorContext context, final String id,
- @Nonnull final ElectionTerm termInformation, final long commitIndex, final long lastApplied,
- @Nonnull final Map<String, String> peerAddresses,
- @Nonnull final ConfigParams configParams, @Nonnull final DataPersistenceProvider persistenceProvider,
- @Nonnull final Consumer<ApplyState> applyStateConsumer, @Nonnull final Logger logger) {
+ final @NonNull ElectionTerm termInformation, final long commitIndex, final long lastApplied,
+ final @NonNull Map<String, String> peerAddresses,
+ final @NonNull ConfigParams configParams, final @NonNull DataPersistenceProvider persistenceProvider,
+ final @NonNull Consumer<ApplyState> applyStateConsumer, final @NonNull Logger logger,
+ final @NonNull Executor executor) {
this.actor = actor;
this.context = context;
this.id = id;
- this.termInformation = Preconditions.checkNotNull(termInformation);
+ this.termInformation = requireNonNull(termInformation);
+ this.executor = requireNonNull(executor);
this.commitIndex = commitIndex;
this.lastApplied = lastApplied;
- this.configParams = Preconditions.checkNotNull(configParams);
- this.persistenceProvider = Preconditions.checkNotNull(persistenceProvider);
- this.log = Preconditions.checkNotNull(logger);
- this.applyStateConsumer = Preconditions.checkNotNull(applyStateConsumer);
+ this.configParams = requireNonNull(configParams);
+ this.persistenceProvider = requireNonNull(persistenceProvider);
+ this.log = requireNonNull(logger);
+ this.applyStateConsumer = requireNonNull(applyStateConsumer);
fileBackedOutputStreamFactory = new FileBackedOutputStreamFactory(
configParams.getFileBackedStreamingThreshold(), configParams.getTempFileDirectory());
- for (Map.Entry<String, String> e: Preconditions.checkNotNull(peerAddresses).entrySet()) {
+ for (Map.Entry<String, String> e : requireNonNull(peerAddresses).entrySet()) {
peerInfoMap.put(e.getKey(), new PeerInfo(e.getKey(), e.getValue(), VotingState.VOTING));
}
}
return actor;
}
+ @Override
+ public final Executor getExecutor() {
+ return executor;
+ }
+
@Override
@SuppressWarnings("checkstyle:IllegalCatch")
public Optional<Cluster> getCluster() {
cluster = Optional.of(Cluster.get(getActorSystem()));
} catch (Exception e) {
// An exception means there's no cluster configured. This will only happen in unit tests.
- log.debug("{}: Could not obtain Cluster: {}", getId(), e);
+ log.debug("{}: Could not obtain Cluster", getId(), e);
cluster = Optional.empty();
}
}
}
void setCurrentBehavior(final RaftActorBehavior behavior) {
- this.currentBehavior = Preconditions.checkNotNull(behavior);
+ this.currentBehavior = requireNonNull(behavior);
}
@Override
}
@Override
- @Nullable
public RaftActorLeadershipTransferCohort getRaftActorLeadershipTransferCohort() {
return leadershipTransferCohort;
}
@Override
@SuppressWarnings("checkstyle:hiddenField")
- public void setRaftActorLeadershipTransferCohort(
- @Nullable final RaftActorLeadershipTransferCohort leadershipTransferCohort) {
+ public void setRaftActorLeadershipTransferCohort(final RaftActorLeadershipTransferCohort leadershipTransferCohort) {
this.leadershipTransferCohort = leadershipTransferCohort;
}
}
* @param callback the Procedure to be notified when persistence is complete (optional).
* @param doAsync if true, the persistent actor can receive subsequent messages to process in between the persist
* call and the execution of the associated callback. If false, subsequent messages are stashed and get
- * delivered after persistence is complete and the associated callback is executed.
+ * delivered after persistence is complete and the associated callback is executed. In either case the
+ * callback is guaranteed to execute in the context of the actor associated with this log.
* @return true if the entry was successfully appended, false otherwise.
*/
boolean appendAndPersist(@Nonnull ReplicatedLogEntry replicatedLogEntry,
return false;
}
- Procedure<ReplicatedLogEntry> persistCallback = persistedLogEntry -> {
- context.getLogger().debug("{}: persist complete {}", context.getId(), persistedLogEntry);
-
- dataSizeSinceLastSnapshot += persistedLogEntry.size();
-
- if (callback != null) {
- callback.apply(persistedLogEntry);
- }
- };
-
if (doAsync) {
- context.getPersistenceProvider().persistAsync(replicatedLogEntry, persistCallback);
+ context.getPersistenceProvider().persistAsync(replicatedLogEntry,
+ entry -> persistCallback(entry, callback));
} else {
- context.getPersistenceProvider().persist(replicatedLogEntry, persistCallback);
+ context.getPersistenceProvider().persist(replicatedLogEntry, entry -> syncPersistCallback(entry, callback));
}
return true;
}
+
+ private void persistCallback(final ReplicatedLogEntry persistedLogEntry,
+ final Procedure<ReplicatedLogEntry> callback) {
+ context.getExecutor().execute(() -> syncPersistCallback(persistedLogEntry, callback));
+ }
+
+ @SuppressWarnings("checkstyle:illegalCatch")
+ private void syncPersistCallback(final ReplicatedLogEntry persistedLogEntry,
+ final Procedure<ReplicatedLogEntry> callback) {
+ context.getLogger().debug("{}: persist complete {}", context.getId(), persistedLogEntry);
+
+ dataSizeSinceLastSnapshot += persistedLogEntry.size();
+
+ if (callback != null) {
+ try {
+ callback.apply(persistedLogEntry);
+ } catch (Exception e) {
+ context.getLogger().error("{}: persist callback failed", context.getId(), e);
+ throw new IllegalStateException("Persist callback failed", e);
+ }
+ }
+ }
}
}
}
+ /**
+ * Message intended for testing to allow triggering persistData via the mailbox.
+ */
+ public static final class TestPersist {
+
+ private final ActorRef actorRef;
+ private final Identifier identifier;
+ private final Payload payload;
+
+ TestPersist(final ActorRef actorRef, final Identifier identifier, final Payload payload) {
+ this.actorRef = actorRef;
+ this.identifier = identifier;
+ this.payload = payload;
+ }
+
+ public ActorRef getActorRef() {
+ return actorRef;
+ }
+
+ public Identifier getIdentifier() {
+ return identifier;
+ }
+
+ public Payload getPayload() {
+ return payload;
+ }
+ }
+
public static class TestRaftActor extends MockRaftActor {
private final ActorRef collectorActor;
return;
}
+ if (message instanceof TestPersist) {
+ persistData(((TestPersist) message).getActorRef(), ((TestPersist) message).getIdentifier(),
+ ((TestPersist) message).getPayload(), false);
+ return;
+ }
+
try {
Predicate drop = dropMessages.get(message.getClass());
if (drop == null || !drop.test(message)) {
protected MockRaftActor(final AbstractBuilder<?, ?> builder) {
super(builder.id, builder.peerAddresses != null ? builder.peerAddresses :
Collections.<String, String>emptyMap(), Optional.fromNullable(builder.config), PAYLOAD_VERSION);
- state = new ArrayList<>();
+ state = Collections.synchronizedList(new ArrayList<>());
this.actorDelegate = mock(RaftActor.class);
this.recoveryCohortDelegate = mock(RaftActorRecoveryCohort.class);
import akka.actor.Props;
import akka.japi.Procedure;
import com.google.common.io.ByteSource;
+import com.google.common.util.concurrent.MoreExecutors;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
public MockRaftActorContext() {
super(null, null, "test", newElectionTerm(), -1, -1, new HashMap<>(),
- new DefaultConfigParamsImpl(), createProvider(), applyState -> { }, LOG);
+ new DefaultConfigParamsImpl(), createProvider(), applyState -> { }, LOG,
+ MoreExecutors.directExecutor());
setReplicatedLog(new MockReplicatedLogBuilder().build());
}
public MockRaftActorContext(final String id, final ActorSystem system, final ActorRef actor) {
super(actor, null, id, newElectionTerm(), -1, -1, new HashMap<>(),
- new DefaultConfigParamsImpl(), createProvider(), applyState -> actor.tell(applyState, actor), LOG);
+ new DefaultConfigParamsImpl(), createProvider(), applyState -> actor.tell(applyState, actor), LOG,
+ MoreExecutors.directExecutor());
this.system = system;
import akka.testkit.TestActorRef;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
RaftActorContextImpl context = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(),
"test", new ElectionTermImpl(createProvider(), "test", LOG), -1, -1,
- peerMap, configParams, createProvider(), applyState -> { }, LOG);
+ peerMap, configParams, createProvider(), applyState -> { }, LOG, MoreExecutors.directExecutor());
assertEquals("getPeerAddress", "peerAddress1", context.getPeerAddress("peer1"));
assertEquals("getPeerAddress", null, context.getPeerAddress("peer2"));
RaftActorContextImpl context = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(),
"test", new ElectionTermImpl(createProvider(), "test", LOG), -1, -1,
Maps.newHashMap(ImmutableMap.<String, String>of("peer1", "peerAddress1")), configParams,
- createProvider(), applyState -> { }, LOG);
+ createProvider(), applyState -> { }, LOG, MoreExecutors.directExecutor());
context.setPeerAddress("peer1", "peerAddress1_1");
assertEquals("getPeerAddress", "peerAddress1_1", context.getPeerAddress("peer1"));
RaftActorContextImpl context = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(),
"self", new ElectionTermImpl(createProvider(), "test", LOG), -1, -1,
Maps.newHashMap(ImmutableMap.<String, String>of("peer1", "peerAddress1")),
- new DefaultConfigParamsImpl(), createProvider(), applyState -> { }, LOG);
+ new DefaultConfigParamsImpl(), createProvider(), applyState -> { }, LOG,
+ MoreExecutors.directExecutor());
context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo("self", false),
new ServerInfo("peer2", true), new ServerInfo("peer3", false))));
import akka.persistence.SnapshotMetadata;
import akka.persistence.SnapshotOffer;
import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.Arrays;
import java.util.Collections;
import org.hamcrest.Description;
context = new RaftActorContextImpl(null, null, localId, new ElectionTermImpl(mockPersistentProvider, "test",
LOG), -1, -1, Collections.<String,String>emptyMap(), configParams,
- mockPersistence, applyState -> { }, LOG);
+ mockPersistence, applyState -> { }, LOG, MoreExecutors.directExecutor());
support = new RaftActorRecoverySupport(context, mockCohort);
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.ByteSource;
+import com.google.common.util.concurrent.MoreExecutors;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
termInfo.update(1, LEADER_ID);
return new RaftActorContextImpl(actor, actor.underlyingActor().getContext(),
id, termInfo, -1, -1, ImmutableMap.of(LEADER_ID, ""), configParams,
- noPersistence, applyState -> actor.tell(applyState, actor), LOG);
+ noPersistence, applyState -> actor.tell(applyState, actor), LOG, MoreExecutors.directExecutor());
}
abstract static class AbstractMockRaftActor extends MockRaftActor {
import akka.persistence.SaveSnapshotFailure;
import akka.persistence.SaveSnapshotSuccess;
import akka.persistence.SnapshotMetadata;
+import com.google.common.util.concurrent.MoreExecutors;
import java.io.OutputStream;
import java.util.Collections;
import java.util.Optional;
context = new RaftActorContextImpl(mockRaftActorRef, null, "test",
new ElectionTermImpl(mockPersistence, "test", LOG), -1, -1, Collections.<String,String>emptyMap(),
- configParams, mockPersistence, applyState -> { }, LOG) {
+ configParams, mockPersistence, applyState -> { }, LOG, MoreExecutors.directExecutor()) {
@Override
public SnapshotManager getSnapshotManager() {
return mockSnapshotManager;
package org.opendaylight.controller.cluster.raft;
+import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.After;
import org.opendaylight.controller.cluster.PersistentDataProvider;
import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
import org.opendaylight.controller.cluster.notifications.RoleChanged;
+import org.opendaylight.controller.cluster.raft.AbstractRaftActorIntegrationTest.TestPersist;
+import org.opendaylight.controller.cluster.raft.AbstractRaftActorIntegrationTest.TestRaftActor;
import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
assertEquals("AppendEntries size", 3, appendEntries.getEntries().size());
}
+
+ @Test
+ @SuppressWarnings("checkstyle:illegalcatch")
+ public void testApplyStateRace() throws Exception {
+ final String leaderId = factory.generateActorId("leader-");
+ final String followerId = factory.generateActorId("follower-");
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ ActorRef mockFollowerActorRef = factory.createActor(MessageCollectorActor.props());
+
+ TestRaftActor.Builder builder = TestRaftActor.newBuilder()
+ .id(leaderId)
+ .peerAddresses(ImmutableMap.of(followerId,
+ mockFollowerActorRef.path().toString()))
+ .config(config)
+ .collectorActor(factory.createActor(
+ MessageCollectorActor.props(), factory.generateActorId(leaderId + "-collector")));
+
+ TestActorRef<MockRaftActor> leaderActorRef = factory.createTestActor(
+ builder.props(), leaderId);
+ MockRaftActor leaderActor = leaderActorRef.underlyingActor();
+ leaderActor.waitForInitializeBehaviorComplete();
+
+ leaderActor.getRaftActorContext().getTermInformation().update(1, leaderId);
+ Leader leader = new Leader(leaderActor.getRaftActorContext());
+ leaderActor.setCurrentBehavior(leader);
+
+ final ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+ leaderActor.setPersistence(new PersistentDataProvider(leaderActor) {
+ @Override
+ public <T> void persistAsync(final T entry, final Procedure<T> procedure) {
+ // needs to be executed from another thread to simulate the persistence actor calling this callback
+ executorService.submit(() -> {
+ try {
+ procedure.apply(entry);
+ } catch (Exception e) {
+ TEST_LOG.info("Fail during async persist callback", e);
+ }
+ }, "persistence-callback");
+ }
+ });
+
+ leader.getFollower(followerId).setNextIndex(0);
+ leader.getFollower(followerId).setMatchIndex(-1);
+
+ // hitting this is flimsy so run multiple times to improve the chance of things
+ // blowing up while breaking actor containment
+ final TestPersist message =
+ new TestPersist(leaderActorRef, new MockIdentifier("1"), new MockPayload("1"));
+ for (int i = 0; i < 100; i++) {
+ leaderActorRef.tell(message, null);
+
+ AppendEntriesReply reply =
+ new AppendEntriesReply(followerId, 1, true, i, 1, (short) 5);
+ leaderActorRef.tell(reply, mockFollowerActorRef);
+ }
+
+ await("Persistence callback.").atMost(5, TimeUnit.SECONDS).until(() -> leaderActor.getState().size() == 100);
+ executorService.shutdown();
+ }
}
import static org.mockito.Mockito.verifyNoMoreInteractions;
import akka.japi.Procedure;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.Collections;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
context = new RaftActorContextImpl(null, null, "test",
new ElectionTermImpl(mockPersistence, "test", LOG), -1, -1, Collections.<String,String>emptyMap(),
- configParams, mockPersistence, applyState -> { }, LOG);
+ configParams, mockPersistence, applyState -> { }, LOG, MoreExecutors.directExecutor());
}
private void verifyPersist(Object message) throws Exception {
import akka.dispatch.Dispatchers;
import akka.testkit.TestActorRef;
import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
Mockito.doReturn(1L).when(mockElectionTerm).getCurrentTerm();
RaftActorContext raftActorContext = new RaftActorContextImpl(candidateActor, candidateActor.actorContext(),
"candidate", mockElectionTerm, -1, -1, setupPeers(4), new DefaultConfigParamsImpl(),
- new NonPersistentDataProvider(Runnable::run), applyState -> { }, LOG);
+ new NonPersistentDataProvider(Runnable::run), applyState -> { }, LOG, MoreExecutors.directExecutor());
raftActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
raftActorContext.getPeerInfo("peer1").setVotingState(VotingState.NON_VOTING);
raftActorContext.getPeerInfo("peer4").setVotingState(VotingState.NON_VOTING);
Code Review / controller.git / commitdiff