RaftActor extends AbstractUntypedPersistentActor { @Deprecated @Override protected void handleCommand(final Object message) { if (serverConfigurationSupport.handleMessage(message, getSender())) { return; } if (snapshotSupport.handleSnapshotMessage(message, getSender())) { return; } if (message instanceof ApplyState) { ApplyState applyState = (ApplyState) message; if (!hasFollowers()) { context.getReplicatedLog().captureSnapshotIfReady(applyState.getReplicatedLogEntry()); context.getSnapshotManager().trimLog(context.getLastApplied()); } possiblyHandleBehaviorMessage(message); } else if (message instanceof ApplyJournalEntries) { ApplyJournalEntries applyEntries = (ApplyJournalEntries) message; LOG.debug("{}: Persisting ApplyJournalEntries with index={}", persistenceId(), applyEntries.getToIndex()); persistence().persistAsync(applyEntries, NoopProcedure.instance()); } else if (message instanceof FindLeader) { getSender().tell( new FindLeaderReply(getLeaderAddress()), getSelf() ); } else if (message instanceof GetOnDemandRaftState) { onGetOnDemandRaftStats(); } else if (message instanceof InitiateCaptureSnapshot) { captureSnapshot(); } else if (message instanceof SwitchBehavior) { switchBehavior((SwitchBehavior) message); } else if (message instanceof LeaderTransitioning) { onLeaderTransitioning((LeaderTransitioning)message); } else if (message instanceof Shutdown) { onShutDown(); } else if (message instanceof Runnable) { ((Runnable)message).run(); } else if (message instanceof NoopPayload) { persistData(null, null, (NoopPayload) message, false); } else if (message instanceof RequestLeadership) { onRequestLeadership((RequestLeadership) message); } else if (!possiblyHandleBehaviorMessage(message)) { handleNonRaftCommand(message); } } protected  RaftActor(final String id, final Map<String, String> peerAddresses,
         final Optional<ConfigParams> configParams, final short payloadVersion); @Override void preStart(); @Override void postStop(); @Override long snapshotSequenceNr(); ActorSelection getLeader(); final DataPersistenceProvider persistence(); void setPersistence(final DataPersistenceProvider provider);  }
@Test public void testRaftActorForwardsToRaftActorSnapshotMessageSupport() { String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); RaftActorSnapshotMessageSupport mockSupport = mock(RaftActorSnapshotMessageSupport.class); TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId) .config(config).snapshotMessageSupport(mockSupport).props()); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); mockRaftActor.waitForRecoveryComplete(); ApplySnapshot applySnapshot = new ApplySnapshot(mock(Snapshot.class)); doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class)); mockRaftActor.handleCommand(applySnapshot); CaptureSnapshotReply captureSnapshotReply = new CaptureSnapshotReply(ByteState.empty(), Optional.empty()); doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class)); mockRaftActor.handleCommand(captureSnapshotReply); SaveSnapshotSuccess saveSnapshotSuccess = new SaveSnapshotSuccess(new SnapshotMetadata("", 0L, 0L)); doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class)); mockRaftActor.handleCommand(saveSnapshotSuccess); SaveSnapshotFailure saveSnapshotFailure = new SaveSnapshotFailure(new SnapshotMetadata("", 0L, 0L), new Throwable()); doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class)); mockRaftActor.handleCommand(saveSnapshotFailure); doReturn(true).when(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT), any(ActorRef.class)); mockRaftActor.handleCommand(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT); doReturn(true).when(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class)); mockRaftActor.handleCommand(GetSnapshot.INSTANCE); verify(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class)); verify(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class)); verify(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class)); verify(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class)); verify(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT), any(ActorRef.class)); verify(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class)); }
@SuppressWarnings("unchecked") @Test public void testApplyJournalEntriesCallsDataPersistence() throws Exception { String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(), config, dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); mockRaftActor.waitForInitializeBehaviorComplete(); mockRaftActor.waitUntilLeader(); mockRaftActor.handleCommand(new ApplyJournalEntries(10)); verify(dataPersistenceProvider).persistAsync(any(ApplyJournalEntries.class), any(Procedure.class)); }