Merge branch 'ecchronos-4.0'

CHANGES.md

@@ -18,10 +18,15 @@
 
 ### Merged from 1.0
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 
 ## Version 4.0.5 (Not yet released)
 
+### Merged from 1.0
+
+* Skip unnecessary reads from repair history - Issue #548
+
 ## Version 4.0.4
 
 ### Merged from 1.0
@@ -74,6 +79,7 @@
 
 ### Merged from 1.0
 
+* Skip unnecessary reads from repair history - Issue #548
 * Fix repair job priority - Issue #515
 
 ## Version 3.0.0
@@ -94,6 +100,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
 
@@ -216,6 +223,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
@@ -226,6 +234,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
 
@@ -263,6 +272,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

@@ -34,13 +34,15 @@ 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 final long myEstimatedRepairTime;
 
     private RepairStateSnapshot(final Builder builder)
     {
         myLastCompletedAt = builder.myLastCompletedAt;
+        myCreatedAt = builder.myCreatedAt;
         myReplicaRepairGroup = builder.myReplicaRepairGroup;
         myVnodeRepairStates = builder.myVnodeRepairStates;
         myEstimatedRepairTime = myVnodeRepairStates.getRepairTime();
@@ -60,6 +62,15 @@ public long getRemainingRepairTime(final long now, final long repairIntervalMs)
         return sum;
     }
 
+    /**
+     * 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.
      *
@@ -119,6 +130,7 @@ public static Builder newBuilder()
     public static class Builder
     {
         private Long myLastCompletedAt;
+        private long myCreatedAt = System.currentTimeMillis();
         private ImmutableList<ReplicaRepairGroup> myReplicaRepairGroup;
         private VnodeRepairStates myVnodeRepairStates;
 
@@ -158,6 +170,18 @@ public Builder withVnodeRepairStates(final VnodeRepairStates vnodeRepairStates)
             return this;
         }
 
+        /**
+         * Build repair state snapshot with created at timestamp.
+         *
+         * @param createdAt The created at timestamp.
+         * @return Builder
+         */
+        public Builder withCreatedAt(final long createdAt)
+        {
+            myCreatedAt = createdAt;
+            return this;
+        }
+
         /**
          * Build repair state snapshot.
          *

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

@@ -69,10 +69,10 @@ public VnodeRepairStates calculateNewState(final TableReference tableReference,
         }
         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;
@@ -27,6 +28,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -423,12 +425,79 @@ public void testWithHistoryAndPreviousAfterScaleOut() throws UnknownHostExceptio
                 newState(range(5, 0), 1236L, 1237L));
     }
 
+    @Test
+    public void testWithHistoryAndPreviousOnlyIteratesOverDiff() throws UnknownHostException
+    {
+        DriverNode node1 = withNode("127.0.0.1");
+        DriverNode 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<DriverNode> replicas = ImmutableSet.of(node1, node2);
+
+        Map<LongTokenRange, ImmutableSet<DriverNode>> tokenToHostMap = new HashMap<>();
+        tokenToHostMap.put(longTokenRange1, replicas);
+        tokenToHostMap.put(longTokenRange2, replicas);
+
+        long range1RepairedAt = 1;
+        long range2RepairedAt = 2;
+        RepairEntry repairEntry1 = new RepairEntry(longTokenRange1, range1RepairedAt, range1RepairedAt, replicas, "SUCCESS");
+        RepairEntry repairEntry2 = new RepairEntry(longTokenRange2, range2RepairedAt, 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());
+
+        assertNewVnodeStates(newState(range(1, 2), range1RepairedAt, range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt, 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, range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt, range2RepairedAt));
+        List<RepairEntry> secondIterateRepairEntries = new ArrayList<>();
+
+        when(repairHistoryProvider.iterate(eq(TABLE_REFERENCE), any(long.class), eq(firstSnapshotCreatedAt),
+                any(Predicate.class))).thenReturn(secondIterateRepairEntries.iterator());
+
+        assertNewVnodeStates(firstRepairStateSnapshot, newState(range(1, 2), range1RepairedAt, range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt, 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, range1RepairedAt),
+                newState(range(2, 3), range2RepairedAt, range2RepairedAt));
+        long updateRange1RepairedAt = 4;
+        RepairEntry repairEntry3 = new RepairEntry(longTokenRange1, updateRange1RepairedAt, 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());
+        assertNewVnodeStates(secondRepairStateSnapshot, newState(range(1, 2), updateRange1RepairedAt, updateRange1RepairedAt),
+                newState(range(2, 3), range2RepairedAt, 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();
     }