Review feedback /1

Author: iamaleksey

src/java/org/apache/cassandra/replication/CoordinatorLog.java

@@ -31,6 +31,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 import org.agrona.collections.Int2ObjectHashMap;
@@ -387,7 +388,7 @@ private long nextSequenceId()
                 int prevTimestamp = MutationId.timestamp(prev);
 
                 // int overflow
-                if (prevOffset == Integer.MAX_VALUE)
+                if (prevOffset == MAX_OFFSET)
                     return -1;
 
                 int nextOffset = prevOffset + 1;
@@ -517,4 +518,13 @@ void deleteFromSystemTable()
     {
         executeInternal(DELETE_QUERY, keyspace, range.left.toString(), range.right.toString(), logId.hostId, logId.hostLogId);
     }
+
+    @VisibleForTesting
+    static void overrideMaxOffsetForTesting(int nexMaxOffset)
+    {
+        MAX_OFFSET = nexMaxOffset;
+    }
+    // don't make volatile unless it genuinely is an issue for some test,
+    // otherwise it should be *fine* as is, and slight overkill to make volatile
+    private static int MAX_OFFSET = Integer.MAX_VALUE;
 }

src/java/org/apache/cassandra/replication/MutationTrackingService.java

@@ -306,10 +306,7 @@ public boolean isDurablyReconciled(ImmutableCoordinatorLogOffsets logOffsets)
         {
             Shard shard = log2ShardMap.get(new CoordinatorLogId(logId));
             if (shard == null)
-            {
-                logger.debug("Could not find shard for logId {}", logId);
-                return false;
-            }
+                throw new IllegalStateException("Could not find shard for logId " + logId);
 
             if (!shard.isDurablyReconciled(logId, logOffsets))
                 return false;

src/java/org/apache/cassandra/replication/Node2OffsetsMap.java

@@ -33,11 +33,11 @@
  */
 public class Node2OffsetsMap
 {
-    private final Int2ObjectHashMap<Offsets.Mutable> offsetstMap;
+    private final Int2ObjectHashMap<Offsets.Mutable> offsetsMap;
 
     public Node2OffsetsMap()
     {
-        offsetstMap = new Int2ObjectHashMap<>(8, 0.65f, false);
+        offsetsMap = new Int2ObjectHashMap<>(8, 0.65f, false);
     }
 
     public static Node2OffsetsMap forParticipants(CoordinatorLogId logId, Participants participants)
@@ -50,21 +50,21 @@ public static Node2OffsetsMap forParticipants(CoordinatorLogId logId, Participan
 
     void set(int node, Offsets.Mutable offsets)
     {
-        offsetstMap.put(node, offsets);
+        offsetsMap.put(node, offsets);
     }
 
     @Nonnull
     Offsets.Mutable get(int node)
     {
-        return Preconditions.checkNotNull(offsetstMap.get(node));
+        return Preconditions.checkNotNull(offsetsMap.get(node));
     }
 
     Offsets.Mutable intersection()
     {
-        Iterator<Offsets.Mutable> iter = offsetstMap.values().iterator();
+        Iterator<Offsets.Mutable> iter = offsetsMap.values().iterator();
 
         Preconditions.checkArgument(iter.hasNext());
-        if (offsetstMap.size() == 1)
+        if (offsetsMap.size() == 1)
             return Offsets.Mutable.copy(iter.next());
 
         Offsets.Mutable intersection = iter.next();
@@ -75,31 +75,31 @@ Offsets.Mutable intersection()
 
     public void add(int node, Offsets offsets)
     {
-        Offsets.Mutable current = offsetstMap.get(node);
+        Offsets.Mutable current = offsetsMap.get(node);
         if (current != null)
             current.addAll(offsets);
         else
-            offsetstMap.put(node, Offsets.Mutable.copy(offsets));
+            offsetsMap.put(node, Offsets.Mutable.copy(offsets));
     }
 
     public void forEach(IntObjConsumer<Offsets.Mutable> consumer)
     {
-        offsetstMap.forEachInt(consumer);
+        offsetsMap.forEachInt(consumer);
     }
 
     public void clear()
     {
-        offsetstMap.clear();
+        offsetsMap.clear();
     }
 
     public int size()
     {
-        return offsetstMap.size();
+        return offsetsMap.size();
     }
 
     void convertToPrimitiveMap(Map<Integer, List<Integer>> into)
     {
-        for (Int2ObjectHashMap<Offsets.Mutable>.EntryIterator iter = offsetstMap.entrySet().iterator(); iter.hasNext();)
+        for (Int2ObjectHashMap<Offsets.Mutable>.EntryIterator iter = offsetsMap.entrySet().iterator(); iter.hasNext();)
         {
             iter.next();
             into.put(iter.getIntKey(), iter.getValue().asList());
@@ -120,6 +120,6 @@ public boolean equals(Object o)
         if (!(o instanceof Node2OffsetsMap))
             return false;
         Node2OffsetsMap that = (Node2OffsetsMap) o;
-        return this.offsetstMap.equals(that.offsetstMap);
+        return this.offsetsMap.equals(that.offsetsMap);
     }
 }

src/java/org/apache/cassandra/replication/Shard.java

@@ -28,6 +28,9 @@
 
 import com.google.common.base.Preconditions;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.agrona.collections.IntArrayList;
 import org.apache.cassandra.cql3.UntypedResultSet;
 import org.apache.cassandra.db.Mutation;
@@ -49,6 +52,8 @@
 
 public class Shard
 {
+    private static final Logger logger = LoggerFactory.getLogger(Shard.class);
+
     final int localNodeId;
     final String keyspace;
     final Range<Token> range;
@@ -103,7 +108,9 @@ synchronized private MutationId maybeRotateLocalLogAndGetNextId()
         MutationId nextId = currentLocalLog.nextId();
         if (nextId != null) // another thread got to rotate before us
             return nextId;
+        CoordinatorLogId oldLogId = currentLocalLog.logId;
         currentLocalLog = createNewPrimayLog();
+        logger.info("Rotated primary log for {}/{} from {} to {}", keyspace, range, oldLogId, currentLocalLog.logId);
         return nextId();
     }