@Test public void testWithHistoryAndPreviousAfterScaleOut() throws UnknownHostException { MockedHost host1 = new MockedHost("127.0.0.1"); MockedHost host2 = new MockedHost("127.0.0.2"); MockedHost host3 = new MockedHost("127.0.0.3"); LongTokenRange longTokenRange1Before = new LongTokenRange(1, 4); LongTokenRange longTokenRange2Before = new LongTokenRange(5, 0); ImmutableSet<Host> replicasBefore = getReplicas(host1, host2); long rangeRepairedAt = 1; RepairHistoryProvider repairHistoryProvider = new MockedRepairHistoryProvider(TABLE_REFERENCE); VnodeRepairStates vnodeRepairStatesBefore = VnodeRepairStates.newBuilder(Arrays.asList( new VnodeRepairState(longTokenRange1Before, replicasBefore, rangeRepairedAt), new VnodeRepairState(longTokenRange2Before, replicasBefore, rangeRepairedAt))) .build(); VnodeRepairStateFactory vnodeRepairStateFactory = new VnodeRepairStateFactoryImpl(mockReplicationState, repairHistoryProvider); LongTokenRange longTokenRange1After = new LongTokenRange(1, 2); LongTokenRange longTokenRange2After = new LongTokenRange(5, 0); ImmutableSet<Host> replicasGroupOneAfter = getReplicas(host1, host3); ImmutableSet<Host> replicasGroupTwoAfter = getReplicas(host1, host2); Map<LongTokenRange, ImmutableSet<Host>> tokenToHostMapAfter = new HashMap<>(); tokenToHostMapAfter.put(longTokenRange1After, replicasGroupOneAfter); tokenToHostMapAfter.put(longTokenRange2After, replicasGroupTwoAfter); when(mockReplicationState.getTokenRangeToReplicas(eq(TABLE_REFERENCE))).thenReturn(tokenToHostMapAfter); VnodeRepairStates expectedVnodeRepairStatesAfter = VnodeRepairStates.newBuilder(Arrays.asList( new VnodeRepairState(longTokenRange1After, replicasGroupOneAfter, rangeRepairedAt), new VnodeRepairState(longTokenRange2After, replicasGroupTwoAfter, rangeRepairedAt))) .build(); RepairStateSnapshot repairStateSnapshot = RepairStateSnapshot.newBuilder() .withLastRepairedAt(rangeRepairedAt) .withVnodeRepairStates(vnodeRepairStatesBefore) .withReplicaRepairGroups(Collections.emptyList()) .build(); VnodeRepairStates actualVnodeRepairStatesAfter = vnodeRepairStateFactory.calculateNewState(TABLE_REFERENCE, repairStateSnapshot); assertThat(actualVnodeRepairStatesAfter).isEqualTo(expectedVnodeRepairStatesAfter); }
@Override public VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous) { Map<LongTokenRange, ImmutableSet<Host>> tokenRangeToReplicaMap = myReplicationState.getTokenRangeToReplicas(tableReference); long lastRepairedAt = previousLastRepairedAt(previous, tokenRangeToReplicaMap); long now = System.currentTimeMillis(); Iterator<RepairEntry> repairEntryIterator; if (lastRepairedAt == VnodeRepairState.UNREPAIRED) { LOG.debug("No last repaired at found for {}, iterating over all repair entries", tableReference); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } else { LOG.debug("Table {} last repaired at {}, iterating repir entries until that time", tableReference, lastRepairedAt); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap); }
VnodeRepairStateFactoryImpl implements VnodeRepairStateFactory { @Override public VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous) { Map<LongTokenRange, ImmutableSet<Host>> tokenRangeToReplicaMap = myReplicationState.getTokenRangeToReplicas(tableReference); long lastRepairedAt = previousLastRepairedAt(previous, tokenRangeToReplicaMap); long now = System.currentTimeMillis(); Iterator<RepairEntry> repairEntryIterator; if (lastRepairedAt == VnodeRepairState.UNREPAIRED) { LOG.debug("No last repaired at found for {}, iterating over all repair entries", tableReference); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } else { LOG.debug("Table {} last repaired at {}, iterating repir entries until that time", tableReference, lastRepairedAt); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap); } }
VnodeRepairStateFactoryImpl implements VnodeRepairStateFactory { @Override public VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous) { Map<LongTokenRange, ImmutableSet<Host>> tokenRangeToReplicaMap = myReplicationState.getTokenRangeToReplicas(tableReference); long lastRepairedAt = previousLastRepairedAt(previous, tokenRangeToReplicaMap); long now = System.currentTimeMillis(); Iterator<RepairEntry> repairEntryIterator; if (lastRepairedAt == VnodeRepairState.UNREPAIRED) { LOG.debug("No last repaired at found for {}, iterating over all repair entries", tableReference); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } else { LOG.debug("Table {} last repaired at {}, iterating repir entries until that time", tableReference, lastRepairedAt); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap); } VnodeRepairStateFactoryImpl(ReplicationState replicationState, RepairHistoryProvider repairHistoryProvider); }
VnodeRepairStateFactoryImpl implements VnodeRepairStateFactory { @Override public VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous) { Map<LongTokenRange, ImmutableSet<Host>> tokenRangeToReplicaMap = myReplicationState.getTokenRangeToReplicas(tableReference); long lastRepairedAt = previousLastRepairedAt(previous, tokenRangeToReplicaMap); long now = System.currentTimeMillis(); Iterator<RepairEntry> repairEntryIterator; if (lastRepairedAt == VnodeRepairState.UNREPAIRED) { LOG.debug("No last repaired at found for {}, iterating over all repair entries", tableReference); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } else { LOG.debug("Table {} last repaired at {}, iterating repir entries until that time", tableReference, lastRepairedAt); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap); } VnodeRepairStateFactoryImpl(ReplicationState replicationState, RepairHistoryProvider repairHistoryProvider); @Override VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous); }
VnodeRepairStateFactoryImpl implements VnodeRepairStateFactory { @Override public VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous) { Map<LongTokenRange, ImmutableSet<Host>> tokenRangeToReplicaMap = myReplicationState.getTokenRangeToReplicas(tableReference); long lastRepairedAt = previousLastRepairedAt(previous, tokenRangeToReplicaMap); long now = System.currentTimeMillis(); Iterator<RepairEntry> repairEntryIterator; if (lastRepairedAt == VnodeRepairState.UNREPAIRED) { LOG.debug("No last repaired at found for {}, iterating over all repair entries", tableReference); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } else { LOG.debug("Table {} last repaired at {}, iterating repir entries until that time", tableReference, lastRepairedAt); repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap)); } return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap); } VnodeRepairStateFactoryImpl(ReplicationState replicationState, RepairHistoryProvider repairHistoryProvider); @Override VnodeRepairStates calculateNewState(TableReference tableReference, RepairStateSnapshot previous);  }