Merge branch 'cassandra-3.11' into trunk

CHANGES.txt

@@ -110,6 +110,7 @@
  * Duplicate the buffer before passing it to analyser in SASI operation (CASSANDRA-13512)
  * Properly evict pstmts from prepared statements cache (CASSANDRA-13641)
 Merged from 3.0:
+ * Fix incorrect [2.1 <- 3.0] serialization of counter cells created in 2.0 (CASSANDRA-13691)
  * Fix invalid writetime for null cells (CASSANDRA-13711)
  * Fix ALTER TABLE statement to atomically propagate changes to the table and its MVs (CASSANDRA-12952)
  * Fixed ambiguous output of nodetool tablestats command (CASSANDRA-13722)
@@ -128,6 +129,7 @@ Merged from 2.2:
 Merged from 2.1:
  * Clone HeartBeatState when building gossip messages. Make its generation/version volatile (CASSANDRA-13700)
 
+
 3.11.0
  * Allow native function calls in CQLSSTableWriter (CASSANDRA-12606)
  * Replace string comparison with regex/number checks in MessagingService test (CASSANDRA-13216)

src/java/org/apache/cassandra/cql3/UpdateParameters.java

@@ -160,12 +160,15 @@ public class UpdateParameters
         // "counter update", which is a temporary state until we run into 'CounterMutation.updateWithCurrentValue()'
         // which does the read-before-write and sets the proper CounterId, clock and updated value.
         //
-        // We thus create a "fake" local shard here. The CounterId/clock used don't matter as this is just a temporary
+        // We thus create a "fake" local shard here. The clock used doesn't matter as this is just a temporary
         // state that will be replaced when processing the mutation in CounterMutation, but the reason we use a 'local'
         // shard is due to the merging rules: if a user includes multiple updates to the same counter in a batch, those
         // multiple updates will be merged in the PartitionUpdate *before* they even reach CounterMutation. So we need
         // such update to be added together, and that's what a local shard gives us.
-        builder.addCell(BufferCell.live(column, timestamp, CounterContext.instance().createLocal(increment)));
+        //
+        // We set counterid to a special value to differentiate between regular pre-2.0 local shards from pre-2.1 era
+        // and "counter update" temporary state cells. Please see CounterContext.createUpdate() for further details.
+        builder.addCell(BufferCell.live(column, timestamp, CounterContext.instance().createUpdate(increment)));
     }
 
     public void setComplexDeletionTime(ColumnMetadata column)

src/java/org/apache/cassandra/db/context/CounterContext.java

@@ -81,6 +81,14 @@ public class CounterContext
     private static final int COUNT_LENGTH = TypeSizes.sizeof(Long.MAX_VALUE);
     private static final int STEP_LENGTH = CounterId.LENGTH + CLOCK_LENGTH + COUNT_LENGTH;
 
+    /*
+     * A special hard-coded value we use for clock ids to differentiate between regular local shards
+     * and 'fake' local shards used to emulate pre-3.0 CounterUpdateCell-s in UpdateParameters.
+     *
+     * Important for handling counter writes and reads during rolling 2.1/2.2 -> 3.0 upgrades.
+     */
+    static final CounterId UPDATE_CLOCK_ID = CounterId.fromInt(0);
+
     private static final Logger logger = LoggerFactory.getLogger(CounterContext.class);
 
     public static enum Relationship
@@ -99,6 +107,35 @@ public class CounterContext
         return LazyHolder.counterContext;
     }
 
+    /**
+     * Creates a counter context with a single local shard with clock id of UPDATE_CLOCK_ID.
+     *
+     * This is only used in a PartitionUpdate until the update has gone through
+     * CounterMutation.apply(), at which point this special local shard will be replaced by a regular global one.
+     * It should never hit commitlog / memtable / disk, but can hit network.
+     *
+     * We use this so that if an update statement has multiple increments of the same counter we properly
+     * add them rather than keeping only one of them.
+     *
+     * NOTE: Before CASSANDRA-13691 we used a regular local shard without a hard-coded clock id value here.
+     * It was problematic, because it was possible to return a false positive, and on read path encode an old counter
+     * cell from 2.0 era with a regular local shard as a counter update, and to break the 2.1 coordinator.
+     */
+    public ByteBuffer createUpdate(long count)
+    {
+        ContextState state = ContextState.allocate(0, 1, 0);
+        state.writeLocal(UPDATE_CLOCK_ID, 1L, count);
+        return state.context;
+    }
+
+    /**
+     * Checks if a context is an update (see createUpdate() for justification).
+     */
+    public boolean isUpdate(ByteBuffer context)
+    {
+        return ContextState.wrap(context).getCounterId().equals(UPDATE_CLOCK_ID);
+    }
+
     /**
      * Creates a counter context with a single global, 2.1+ shard (a result of increment).
      */
@@ -111,12 +148,7 @@ public class CounterContext
 
     /**
      * Creates a counter context with a single local shard.
-     * This is only used in a PartitionUpdate until the update has gone through
-     * CounterMutation.apply(), at which point all the local shard are replaced by
-     * global ones. In other words, local shards should never hit the disk or
-     * memtables. And we use this so that if an update statement has multiple increment
-     * of the same counter we properly add them rather than keeping only one of them.
-     * (this is also used for tests of compatibility with pre-2.1 counters)
+     * For use by tests of compatibility with pre-2.1 counters only.
      */
     public ByteBuffer createLocal(long count)
     {
@@ -681,14 +713,6 @@ public class CounterContext
         return getLocalClockAndCount(context).count;
     }
 
-    /**
-     * Checks if a context is local
-     */
-    public boolean isLocal(ByteBuffer context)
-    {
-        return ContextState.wrap(context).isLocal();
-    }
-
     /**
      * Returns the clock and the count associated with the given counter id, or (0, 0) if no such shard is present.
      */

src/java/org/apache/cassandra/utils/CounterId.java

@@ -46,10 +46,11 @@ public class CounterId implements Comparable<CounterId>
     }
 
     /**
-     * Function for test purposes, do not use otherwise.
      * Pack an int in a valid CounterId so that the resulting ids respects the
      * numerical ordering. Used for creating handcrafted but easy to
      * understand contexts in unit tests (see CounterContextTest).
+     *
+     * Also used to generate a special ID for special-case update contexts (see CounterContext.createUpdate()).
      */
     public static CounterId fromInt(int n)
     {

test/unit/org/apache/cassandra/db/context/CounterContextTest.java

@@ -32,13 +32,14 @@ import org.apache.cassandra.db.context.CounterContext.Relationship;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.CounterId;
 
-import static org.apache.cassandra.db.context.CounterContext.ContextState;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import static org.apache.cassandra.db.context.CounterContext.ContextState;
+
 public class CounterContextTest
 {
     private static final CounterContext cc = new CounterContext();
@@ -542,4 +543,30 @@ public class CounterContextTest
         assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(15)));
         assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(20)));
     }
+
+    @Test // see CASSANDRA-13691
+    public void testCounterUpdate()
+    {
+        /*
+         * a context with just one 'update' shard - a local shard with a hardcoded value of CounterContext.UPDATE_CLOCK_ID
+         */
+
+        ByteBuffer updateContext = CounterContext.instance().createUpdate(10L);
+
+        assertEquals(ClockAndCount.create(1L, 10L), cc.getClockAndCountOf(updateContext, CounterContext.UPDATE_CLOCK_ID));
+        assertTrue(cc.isUpdate(updateContext));
+
+
+        /*
+         * a context with a regular local shard sorting first and a couple others in it - should *not* be identified as an update
+         */
+
+        ContextState notUpdateContextState = ContextState.allocate(1, 1, 1);
+        notUpdateContextState.writeLocal( CounterId.fromInt(1), 1L, 10L);
+        notUpdateContextState.writeRemote(CounterId.fromInt(2), 1L, 10L);
+        notUpdateContextState.writeGlobal(CounterId.fromInt(3), 1L, 10L);
+        ByteBuffer notUpdateContext = notUpdateContextState.context;
+
+        assertFalse(cc.isUpdate(notUpdateContext));
+    }
 }