import akka.actor.PoisonPill;
import akka.actor.Terminated;
import akka.dispatch.Dispatchers;
+import akka.dispatch.Mailboxes;
import akka.testkit.JavaTestKit;
import akka.testkit.TestActorRef;
import akka.util.Timeout;
import com.google.common.base.Stopwatch;
-import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Uninterruptibles;
+import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Predicate;
+import org.apache.commons.lang3.SerializationUtils;
import org.junit.After;
+import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
public void handleCommand(Object message) {
if (message instanceof MockPayload) {
MockPayload payload = (MockPayload) message;
- super.persistData(collectorActor, new MockIdentifier(payload.toString()), payload);
+ super.persistData(collectorActor, new MockIdentifier(payload.toString()), payload, false);
return;
}
if (message instanceof ServerConfigurationPayload) {
- super.persistData(collectorActor, new MockIdentifier("serverConfig"), (Payload) message);
+ super.persistData(collectorActor, new MockIdentifier("serverConfig"), (Payload) message, false);
return;
}
@Override
@SuppressWarnings("checkstyle:IllegalCatch")
- public void createSnapshot(ActorRef actorRef) {
- try {
- actorRef.tell(new CaptureSnapshotReply(RaftActorTest.fromObject(getState()).toByteArray()), actorRef);
- } catch (Exception e) {
- Throwables.propagate(e);
+ public void createSnapshot(ActorRef actorRef, Optional<OutputStream> installSnapshotStream) {
+ MockSnapshotState snapshotState = new MockSnapshotState(new ArrayList<>(getState()));
+ if (installSnapshotStream.isPresent()) {
+ SerializationUtils.serialize(snapshotState, installSnapshotStream.get());
}
+
+ actorRef.tell(new CaptureSnapshotReply(snapshotState, installSnapshotStream), actorRef);
}
public ActorRef collectorActor() {
protected long currentTerm;
protected int snapshotBatchCount = 4;
+ protected int snapshotChunkSize = SNAPSHOT_CHUNK_SIZE;
protected List<MockPayload> expSnapshotState = new ArrayList<>();
configParams.setSnapshotBatchCount(snapshotBatchCount);
configParams.setSnapshotDataThresholdPercentage(70);
configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
- configParams.setSnapshotChunkSize(SNAPSHOT_CHUNK_SIZE);
+ configParams.setSnapshotChunkSize(snapshotChunkSize);
return configParams;
}
InvalidActorNameException lastEx = null;
for (int i = 0; i < 10; i++) {
try {
- return factory.createTestActor(builder.props().withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ return factory.createTestActor(builder.props().withDispatcher(Dispatchers.DefaultDispatcherId())
+ .withMailbox(Mailboxes.DefaultMailboxId()), id);
} catch (InvalidActorNameException e) {
lastEx = e;
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
msg -> msg.getToIndex() == expIndex);
}
- @SuppressWarnings("unchecked")
protected void verifySnapshot(String prefix, Snapshot snapshot, long lastAppliedTerm,
long lastAppliedIndex, long lastTerm, long lastIndex)
throws Exception {
assertEquals(prefix + " Snapshot getLastTerm", lastTerm, snapshot.getLastTerm());
assertEquals(prefix + " Snapshot getLastIndex", lastIndex, snapshot.getLastIndex());
- List<Object> actualState = (List<Object>)MockRaftActor.toObject(snapshot.getState());
+ List<Object> actualState = ((MockSnapshotState)snapshot.getState()).getState();
assertEquals(String.format("%s Snapshot getState size. Expected %s: . Actual: %s", prefix, expSnapshotState,
actualState), expSnapshotState.size(), actualState.size());
for (int i = 0; i < expSnapshotState.size(); i++) {