@Test public void testUpdate() { SyncStatusTracker tracker = new SyncStatusTracker(listener, "commit-tracker", 10); tracker.update("leader-1", 100, 99); FollowerInitialSyncUpStatus status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertEquals(false, status.isInitialSyncDone()); MessageCollectorActor.clearMessages(listener); tracker.update("leader-1", 105, 101); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertEquals(true, status.isInitialSyncDone()); MessageCollectorActor.clearMessages(listener); tracker.update("leader-1", 108, 101); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertNull("No status message should be received", status); tracker.update("leader-1", 150, 101); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertNotNull("No sync status message was received", status); assertEquals(false, status.isInitialSyncDone()); MessageCollectorActor.clearMessages(listener); tracker.update("leader-1", 150, 125); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertNull("No status message should be received", status); tracker.update("leader-1", 160, 155); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertEquals(true, status.isInitialSyncDone()); MessageCollectorActor.clearMessages(listener); tracker.update("leader-2", 160, 155); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertEquals(false, status.isInitialSyncDone()); MessageCollectorActor.clearMessages(listener); tracker.update("leader-2", 160, 159); status = MessageCollectorActor.getFirstMatching(listener, FollowerInitialSyncUpStatus.class); assertNull("No status message should be received", status); }
public void update(final String leaderId, final long leaderCommit, final long commitIndex) { requireNonNull(leaderId, "leaderId should not be null"); if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) { LOG.debug("{}: Last sync leader does not match current leader {}, need to catch up to {}", id, leaderId, leaderCommit); changeSyncStatus(NOT_IN_SYNC, true); syncTarget = new LeaderInfo(leaderId, leaderCommit); return; } final long lag = leaderCommit - commitIndex; if (lag > syncThreshold) { LOG.debug("{}: Lagging {} entries behind leader {}", id, lag, leaderId); changeSyncStatus(NOT_IN_SYNC, false); } else if (commitIndex >= syncTarget.minimumCommitIndex) { LOG.debug("{}: Lagging {} entries behind leader {} and reached {} (of expected {})", id, lag, leaderId, commitIndex, syncTarget.minimumCommitIndex); changeSyncStatus(IN_SYNC, false); } }
SyncStatusTracker { public void update(final String leaderId, final long leaderCommit, final long commitIndex) { requireNonNull(leaderId, "leaderId should not be null"); if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) { LOG.debug("{}: Last sync leader does not match current leader {}, need to catch up to {}", id, leaderId, leaderCommit); changeSyncStatus(NOT_IN_SYNC, true); syncTarget = new LeaderInfo(leaderId, leaderCommit); return; } final long lag = leaderCommit - commitIndex; if (lag > syncThreshold) { LOG.debug("{}: Lagging {} entries behind leader {}", id, lag, leaderId); changeSyncStatus(NOT_IN_SYNC, false); } else if (commitIndex >= syncTarget.minimumCommitIndex) { LOG.debug("{}: Lagging {} entries behind leader {} and reached {} (of expected {})", id, lag, leaderId, commitIndex, syncTarget.minimumCommitIndex); changeSyncStatus(IN_SYNC, false); } } }
SyncStatusTracker { public void update(final String leaderId, final long leaderCommit, final long commitIndex) { requireNonNull(leaderId, "leaderId should not be null"); if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) { LOG.debug("{}: Last sync leader does not match current leader {}, need to catch up to {}", id, leaderId, leaderCommit); changeSyncStatus(NOT_IN_SYNC, true); syncTarget = new LeaderInfo(leaderId, leaderCommit); return; } final long lag = leaderCommit - commitIndex; if (lag > syncThreshold) { LOG.debug("{}: Lagging {} entries behind leader {}", id, lag, leaderId); changeSyncStatus(NOT_IN_SYNC, false); } else if (commitIndex >= syncTarget.minimumCommitIndex) { LOG.debug("{}: Lagging {} entries behind leader {} and reached {} (of expected {})", id, lag, leaderId, commitIndex, syncTarget.minimumCommitIndex); changeSyncStatus(IN_SYNC, false); } } SyncStatusTracker(final ActorRef actor, final String id, final long syncThreshold); }
SyncStatusTracker { public void update(final String leaderId, final long leaderCommit, final long commitIndex) { requireNonNull(leaderId, "leaderId should not be null"); if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) { LOG.debug("{}: Last sync leader does not match current leader {}, need to catch up to {}", id, leaderId, leaderCommit); changeSyncStatus(NOT_IN_SYNC, true); syncTarget = new LeaderInfo(leaderId, leaderCommit); return; } final long lag = leaderCommit - commitIndex; if (lag > syncThreshold) { LOG.debug("{}: Lagging {} entries behind leader {}", id, lag, leaderId); changeSyncStatus(NOT_IN_SYNC, false); } else if (commitIndex >= syncTarget.minimumCommitIndex) { LOG.debug("{}: Lagging {} entries behind leader {} and reached {} (of expected {})", id, lag, leaderId, commitIndex, syncTarget.minimumCommitIndex); changeSyncStatus(IN_SYNC, false); } } SyncStatusTracker(final ActorRef actor, final String id, final long syncThreshold); void update(final String leaderId, final long leaderCommit, final long commitIndex); }
SyncStatusTracker { public void update(final String leaderId, final long leaderCommit, final long commitIndex) { requireNonNull(leaderId, "leaderId should not be null"); if (syncTarget == null || !leaderId.equals(syncTarget.leaderId)) { LOG.debug("{}: Last sync leader does not match current leader {}, need to catch up to {}", id, leaderId, leaderCommit); changeSyncStatus(NOT_IN_SYNC, true); syncTarget = new LeaderInfo(leaderId, leaderCommit); return; } final long lag = leaderCommit - commitIndex; if (lag > syncThreshold) { LOG.debug("{}: Lagging {} entries behind leader {}", id, lag, leaderId); changeSyncStatus(NOT_IN_SYNC, false); } else if (commitIndex >= syncTarget.minimumCommitIndex) { LOG.debug("{}: Lagging {} entries behind leader {} and reached {} (of expected {})", id, lag, leaderId, commitIndex, syncTarget.minimumCommitIndex); changeSyncStatus(IN_SYNC, false); } } SyncStatusTracker(final ActorRef actor, final String id, final long syncThreshold); void update(final String leaderId, final long leaderCommit, final long commitIndex);  }