Merge branch 'ecchronos-1.2' into ecchronos-2.0

CHANGES.md

@@ -4,6 +4,7 @@
 
 ### Merged from 1.0
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 * Fix malformed IPv6 for JMX - Issue #306
 
@@ -126,6 +127,7 @@
 
 ### Merged from 1.0
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 * Fix malformed IPv6 for JMX - Issue #306
 * Step karaf to 4.2.8
@@ -136,6 +138,7 @@
 
 #### Merged from 1.0
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 * Fix malformed IPv6 for JMX - Issue #306
 
@@ -173,6 +176,7 @@
 
 ## Version 1.0.8 (Not yet released)
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 * Fix malformed IPv6 for JMX - Issue #306
 * Step karaf to 4.2.8

core/src/main/java/com/ericsson/bss/cassandra/ecchronos/core/repair/state/RepairStateSnapshot.java

@@ -33,18 +33,29 @@ public class RepairStateSnapshot
 {
     private final boolean canRepair;
     private final long myLastCompletedAt;
+    private final long myCreatedAt;
     private final ImmutableList<ReplicaRepairGroup> myReplicaRepairGroup;
     private final VnodeRepairStates myVnodeRepairStates;
 
     private RepairStateSnapshot(Builder builder)
     {
         myLastCompletedAt = builder.myLastCompletedAt;
+        myCreatedAt = builder.myCreatedAt;
         myReplicaRepairGroup = builder.myReplicaRepairGroup;
         myVnodeRepairStates = builder.myVnodeRepairStates;
 
         canRepair = !myReplicaRepairGroup.isEmpty();
     }
 
+    /**
+     * Get the time this snapshot was created.
+     * @return The time this snapshot was created.
+     */
+    public long getCreatedAt()
+    {
+        return myCreatedAt;
+    }
+
     /**
      * Check if a repair can be performed based on the current state.
      *
@@ -94,6 +105,7 @@ public static Builder newBuilder()
     public static class Builder
     {
         private Long myLastCompletedAt;
+        private long myCreatedAt = System.currentTimeMillis();
         private ImmutableList<ReplicaRepairGroup> myReplicaRepairGroup;
         private VnodeRepairStates myVnodeRepairStates;
 
@@ -115,6 +127,12 @@ public Builder withVnodeRepairStates(VnodeRepairStates vnodeRepairStates)
             return this;
         }
 
+        public Builder withCreatedAt(long createdAt)
+        {
+            myCreatedAt = createdAt;
+            return this;
+        }
+
         public RepairStateSnapshot build()
         {
             return new RepairStateSnapshot(this);

core/src/main/java/com/ericsson/bss/cassandra/ecchronos/core/repair/state/VnodeRepairStateFactoryImpl.java

@@ -60,8 +60,8 @@ public VnodeRepairStates calculateNewState(TableReference tableReference, Repair
         }
         else
         {
-            LOG.debug("Table {} last repaired at {}, iterating repair entries until that time", tableReference, lastRepairedAt);
-            repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, lastRepairedAt, (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap));
+            LOG.debug("Table {} snapshot created at {}, iterating repir entries until that time", tableReference, previous.getCreatedAt());
+            repairEntryIterator = myRepairHistoryProvider.iterate(tableReference, now, previous.getCreatedAt(), (repairEntry) -> acceptRepairEntries(repairEntry, tokenRangeToReplicaMap));
         }
 
         return generateVnodeRepairStates(lastRepairedAt, previous, repairEntryIterator, tokenRangeToReplicaMap);

core/src/test/java/com/ericsson/bss/cassandra/ecchronos/core/repair/state/TestVnodeRepairStateFactoryImpl.java

@@ -16,6 +16,7 @@
 
 import static com.ericsson.bss.cassandra.ecchronos.core.MockTableReferenceFactory.tableReference;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -26,6 +27,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -252,12 +254,79 @@ public void testWithHistoryAndPreviousAfterScaleOut() throws UnknownHostExceptio
                 newState(range(5, 0), 1234L));
     }
 
+    @Test
+    public void testWithHistoryAndPreviousOnlyIteratesOverDiff() throws UnknownHostException
+    {
+        Node node1 = withNode("127.0.0.1");
+        Node node2 = withNode("127.0.0.2");
+
+        LongTokenRange longTokenRange1 = range(1, 2);
+        LongTokenRange longTokenRange2 = range(2, 3);
+        withRange(longTokenRange1, node1, node2);
+        withRange(longTokenRange2, node1, node2);
+        ImmutableSet<Node> replicas = ImmutableSet.of(node1, node2);
+
+        Map<LongTokenRange, ImmutableSet<Node>> tokenToHostMap = new HashMap<>();
+        tokenToHostMap.put(longTokenRange1, replicas);
+        tokenToHostMap.put(longTokenRange2, replicas);
+
+        long range1RepairedAt = 1;
+        long range2RepairedAt = 2;
+        RepairEntry repairEntry1 = new RepairEntry(longTokenRange1, range1RepairedAt, replicas, "SUCCESS");
+        RepairEntry repairEntry2 = new RepairEntry(longTokenRange2, range2RepairedAt, replicas, "SUCCESS");
+        List<RepairEntry> firstIterateRepairEntries = new ArrayList<>();
+        firstIterateRepairEntries.add(repairEntry1);
+        firstIterateRepairEntries.add(repairEntry2);
+
+        when(mockReplicationState.getTokenRangeToReplicas(eq(TABLE_REFERENCE))).thenReturn(tokenToHostMap);
+        repairHistoryProvider = mock(RepairHistoryProvider.class);
+        when(repairHistoryProvider.iterate(eq(TABLE_REFERENCE), any(long.class), any(Predicate.class))).thenReturn(
+                firstIterateRepairEntries.iterator());
+
+        assertVnodeStates(newState(range(1, 2), range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt));
+
+        // Check that vnodes keep their states from old snapshot even if iterator is empty
+        long firstSnapshotCreatedAt = 3;
+        RepairStateSnapshot firstRepairStateSnapshot = snapshot(range1RepairedAt, firstSnapshotCreatedAt,
+                newState(range(1, 2), range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt));
+        List<RepairEntry> secondIterateRepairEntries = new ArrayList<>();
+
+        when(repairHistoryProvider.iterate(eq(TABLE_REFERENCE), any(long.class), eq(firstSnapshotCreatedAt),
+                any(Predicate.class))).thenReturn(secondIterateRepairEntries.iterator());
+
+        assertVnodeStates(firstRepairStateSnapshot, newState(range(1, 2), range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt));
+
+        // Check that vnodes get updated for the new repair entries and old are kept from old snapshot
+        long secondSnapshotCreatedAt = 5;
+        RepairStateSnapshot secondRepairStateSnapshot = snapshot(range1RepairedAt, secondSnapshotCreatedAt,
+                newState(range(1, 2), range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt));
+        long updateRange1RepairedAt = 4;
+        RepairEntry repairEntry3 = new RepairEntry(longTokenRange1, updateRange1RepairedAt, replicas, "SUCCESS");
+        List<RepairEntry> thirdIterateRepairEntries = new ArrayList<>();
+        thirdIterateRepairEntries.add(repairEntry3);
+
+        when(repairHistoryProvider.iterate(eq(TABLE_REFERENCE), any(long.class), eq(secondSnapshotCreatedAt),
+                any(Predicate.class))).thenReturn(thirdIterateRepairEntries.iterator());
+        assertVnodeStates(secondRepairStateSnapshot, newState(range(1, 2), updateRange1RepairedAt),
+                newState(range(2, 3), range2RepairedAt));
+    }
+
     private RepairStateSnapshot snapshot(long repairedAt, VnodeRepairState... states)
+    {
+        return snapshot(repairedAt, repairedAt, states);
+    }
+
+    private RepairStateSnapshot snapshot(long repairedAt, long createdAt, VnodeRepairState... states)
     {
         return RepairStateSnapshot.newBuilder()
                 .withLastCompletedAt(repairedAt)
                 .withReplicaRepairGroups(Collections.emptyList())
                 .withVnodeRepairStates(vnodeRepairStates(states))
+                .withCreatedAt(createdAt)
                 .build();
     }