content: /*
   *
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   *
   */
  package org.apache.cassandra.db.context;
  
  import java.nio.ByteBuffer;
  
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.Util;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.ClockAndCount;
  import org.apache.cassandra.db.commitlog.CommitLog;
  import org.apache.cassandra.db.context.CounterContext.Relationship;
  import org.apache.cassandra.db.marshal.ByteBufferAccessor;
  import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
  import org.apache.cassandra.tcm.ClusterMetadataService;
  import org.apache.cassandra.tcm.StubClusterMetadataService;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.CounterId;
  import org.apache.cassandra.utils.FBUtilities;
  
  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();
  
      private static final int headerSizeLength = 2;
      private static final int headerEltLength = 2;
      private static final int idLength = 16;
      private static final int clockLength = 8;
      private static final int countLength = 8;
      private static final int stepLength = idLength + clockLength + countLength;
  
      @BeforeClass
      public static void setupDD()
      {
          DatabaseDescriptor.daemonInitialization();
          CommitLog.instance.start();
          ClusterMetadataService.unsetInstance();
          ClusterMetadataService.setInstance(StubClusterMetadataService.forTesting());
          ClusterMetadataTestHelper.register(FBUtilities.getBroadcastAddressAndPort());
      }
  
      @Test
      public void testAllocate()
      {
          ContextState allGlobal = ContextState.allocate(3, 0, 0);
          assertEquals(headerSizeLength + 3 * headerEltLength + 3 * stepLength, allGlobal.context.remaining());
  
          ContextState allLocal = ContextState.allocate(0, 3, 0);
          assertEquals(headerSizeLength + 3 * headerEltLength + 3 * stepLength, allLocal.context.remaining());
  
          ContextState allRemote = ContextState.allocate(0, 0, 3);
          assertEquals(headerSizeLength + 3 * stepLength, allRemote.context.remaining());
  
          ContextState mixed = ContextState.allocate(1, 1, 1);
          assertEquals(headerSizeLength + 2 * headerEltLength + 3 * stepLength, mixed.context.remaining());
      }
  
      @Test
      public void testDiff()
      {
          ContextState left;
          ContextState right;
  
          // equality: equal nodes, all counts same
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 3L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
          right = ContextState.wrap(ByteBufferUtil.clone(left.context));
  
          assertEquals(Relationship.EQUAL, cc.diff(left.context, right.context));
  
          // greater than: left has superset of nodes (counts equal)
          left = ContextState.allocate(0, 0, 4);
          left.writeRemote(CounterId.fromInt(3),  3L, 0L);
          left.writeRemote(CounterId.fromInt(6),  2L, 0L);
          left.writeRemote(CounterId.fromInt(9),  1L, 0L);
          left.writeRemote(CounterId.fromInt(12), 0L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 3L, 0L);
          right.writeRemote(CounterId.fromInt(6), 2L, 0L);
          right.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          assertEquals(Relationship.GREATER_THAN, cc.diff(left.context, right.context));
  
          // less than: left has subset of nodes (counts equal)
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 3L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 4);
          right.writeRemote(CounterId.fromInt(3),  3L, 0L);
          right.writeRemote(CounterId.fromInt(6),  2L, 0L);
          right.writeRemote(CounterId.fromInt(9),  1L, 0L);
          right.writeRemote(CounterId.fromInt(12), 0L, 0L);
  
          assertEquals(Relationship.LESS_THAN, cc.diff(left.context, right.context));
  
          // greater than: equal nodes, but left has higher counts
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 3L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 3L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 3L, 0L);
          right.writeRemote(CounterId.fromInt(6), 2L, 0L);
          right.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          assertEquals(Relationship.GREATER_THAN, cc.diff(left.context, right.context));
  
          // less than: equal nodes, but right has higher counts
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 3L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 3L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 3L, 0L);
          right.writeRemote(CounterId.fromInt(6), 9L, 0L);
          right.writeRemote(CounterId.fromInt(9), 3L, 0L);
  
          assertEquals(Relationship.LESS_THAN, cc.diff(left.context, right.context));
  
          // disjoint: right and left have disjoint node sets
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 1L, 0L);
          left.writeRemote(CounterId.fromInt(4), 1L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 1L, 0L);
          right.writeRemote(CounterId.fromInt(6), 1L, 0L);
          right.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 1L, 0L);
          left.writeRemote(CounterId.fromInt(4), 1L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(2),  1L, 0L);
          right.writeRemote(CounterId.fromInt(6),  1L, 0L);
          right.writeRemote(CounterId.fromInt(12), 1L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          // disjoint: equal nodes, but right and left have higher counts in differing nodes
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 1L, 0L);
          left.writeRemote(CounterId.fromInt(6), 3L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 1L, 0L);
          right.writeRemote(CounterId.fromInt(6), 1L, 0L);
          right.writeRemote(CounterId.fromInt(9), 5L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 2L, 0L);
          left.writeRemote(CounterId.fromInt(6), 3L, 0L);
          left.writeRemote(CounterId.fromInt(9), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 1L, 0L);
          right.writeRemote(CounterId.fromInt(6), 9L, 0L);
          right.writeRemote(CounterId.fromInt(9), 5L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          // disjoint: left has more nodes, but lower counts
          left = ContextState.allocate(0, 0, 4);
          left.writeRemote(CounterId.fromInt(3),  2L, 0L);
          left.writeRemote(CounterId.fromInt(6),  3L, 0L);
          left.writeRemote(CounterId.fromInt(9),  1L, 0L);
          left.writeRemote(CounterId.fromInt(12), 1L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 4L, 0L);
          right.writeRemote(CounterId.fromInt(6), 9L, 0L);
          right.writeRemote(CounterId.fromInt(9), 5L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          // disjoint: left has less nodes, but higher counts
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 5L, 0L);
          left.writeRemote(CounterId.fromInt(6), 3L, 0L);
          left.writeRemote(CounterId.fromInt(9), 2L, 0L);
  
          right = ContextState.allocate(0, 0, 4);
          right.writeRemote(CounterId.fromInt(3),  4L, 0L);
          right.writeRemote(CounterId.fromInt(6),  3L, 0L);
          right.writeRemote(CounterId.fromInt(9),  2L, 0L);
          right.writeRemote(CounterId.fromInt(12), 1L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          // disjoint: mixed nodes and counts
          left = ContextState.allocate(0, 0, 3);
          left.writeRemote(CounterId.fromInt(3), 5L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 2L, 0L);
  
          right = ContextState.allocate(0, 0, 4);
          right.writeRemote(CounterId.fromInt(3),  4L, 0L);
          right.writeRemote(CounterId.fromInt(6),  3L, 0L);
          right.writeRemote(CounterId.fromInt(9),  2L, 0L);
          right.writeRemote(CounterId.fromInt(12), 1L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
  
          left = ContextState.allocate(0, 0, 4);
          left.writeRemote(CounterId.fromInt(3), 5L, 0L);
          left.writeRemote(CounterId.fromInt(6), 2L, 0L);
          left.writeRemote(CounterId.fromInt(7), 2L, 0L);
          left.writeRemote(CounterId.fromInt(9), 2L, 0L);
  
          right = ContextState.allocate(0, 0, 3);
          right.writeRemote(CounterId.fromInt(3), 4L, 0L);
          right.writeRemote(CounterId.fromInt(6), 3L, 0L);
          right.writeRemote(CounterId.fromInt(9), 2L, 0L);
  
          assertEquals(Relationship.DISJOINT, cc.diff(left.context, right.context));
      }
  
      @Test
      public void testMerge()
      {
          // note: local counts aggregated; remote counts are reconciled (i.e. take max)
          ContextState left = ContextState.allocate(0, 1, 3);
          left.writeRemote(CounterId.fromInt(1), 1L, 1L);
          left.writeRemote(CounterId.fromInt(2), 2L, 2L);
          left.writeRemote(CounterId.fromInt(4), 6L, 3L);
          left.writeLocal(CounterId.getLocalId(), 7L, 3L);
  
          ContextState right = ContextState.allocate(0, 1, 2);
          right.writeRemote(CounterId.fromInt(4), 4L, 4L);
          right.writeRemote(CounterId.fromInt(5), 5L, 5L);
          right.writeLocal(CounterId.getLocalId(), 2L, 9L);
  
          ByteBuffer merged = cc.merge(left.context, right.context);
          int hd = 4;
  
          assertEquals(hd + 5 * stepLength, merged.remaining());
          // local node id's counts are aggregated
          assertTrue(Util.equalsCounterId(CounterId.getLocalId(), merged, hd + 4 * stepLength));
          assertEquals(9L, merged.getLong(merged.position() + hd + 4 * stepLength + idLength));
          assertEquals(12L,  merged.getLong(merged.position() + hd + 4*stepLength + idLength + clockLength));
  
          // remote node id counts are reconciled (i.e. take max)
          assertTrue(Util.equalsCounterId(CounterId.fromInt(4), merged, hd + 2 * stepLength));
          assertEquals(6L, merged.getLong(merged.position() + hd + 2 * stepLength + idLength));
          assertEquals( 3L,  merged.getLong(merged.position() + hd + 2*stepLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(5), merged, hd + 3 * stepLength));
          assertEquals(5L, merged.getLong(merged.position() + hd + 3 * stepLength + idLength));
          assertEquals( 5L,  merged.getLong(merged.position() + hd + 3*stepLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(2), merged, hd + stepLength));
          assertEquals(2L, merged.getLong(merged.position() + hd + stepLength + idLength));
          assertEquals( 2L,  merged.getLong(merged.position() + hd + stepLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), merged, hd));
          assertEquals( 1L,  merged.getLong(merged.position() + hd + idLength));
          assertEquals( 1L,  merged.getLong(merged.position() + hd + idLength + clockLength));
  
          //
          // Test merging two exclusively global contexts
          //
          left = ContextState.allocate(3, 0, 0);
          left.writeGlobal(CounterId.fromInt(1), 1L, 1L);
          left.writeGlobal(CounterId.fromInt(2), 2L, 2L);
          left.writeGlobal(CounterId.fromInt(3), 3L, 3L);
  
          right = ContextState.allocate(3, 0, 0);
          right.writeGlobal(CounterId.fromInt(3), 6L, 6L);
          right.writeGlobal(CounterId.fromInt(4), 4L, 4L);
          right.writeGlobal(CounterId.fromInt(5), 5L, 5L);
  
          merged = cc.merge(left.context, right.context);
          assertEquals(headerSizeLength + 5 * headerEltLength + 5 * stepLength, merged.remaining());
          assertEquals(18L, cc.total(merged, ByteBufferAccessor.instance));
          assertEquals(5, merged.getShort(merged.position()));
  
          int headerLength = headerSizeLength + 5 * headerEltLength;
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), merged, headerLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + idLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + idLength + clockLength));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(2), merged, headerLength + stepLength));
          assertEquals(2L, merged.getLong(merged.position() + headerLength + stepLength + idLength));
          assertEquals(2L, merged.getLong(merged.position() + headerLength + stepLength + idLength + clockLength));
          // pick the global shard with the largest clock
          assertTrue(Util.equalsCounterId(CounterId.fromInt(3), merged, headerLength + 2 * stepLength));
          assertEquals(6L, merged.getLong(merged.position() + headerLength + 2 * stepLength + idLength));
          assertEquals(6L, merged.getLong(merged.position() + headerLength + 2 * stepLength + idLength + clockLength));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(4), merged, headerLength + 3 * stepLength));
          assertEquals(4L, merged.getLong(merged.position() + headerLength + 3 * stepLength + idLength));
          assertEquals(4L, merged.getLong(merged.position() + headerLength + 3 * stepLength + idLength + clockLength));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(5), merged, headerLength + 4 * stepLength));
          assertEquals(5L, merged.getLong(merged.position() + headerLength + 4 * stepLength + idLength));
          assertEquals(5L, merged.getLong(merged.position() + headerLength + 4 * stepLength + idLength + clockLength));
  
          //
          // Test merging two global contexts w/ 'invalid shards'
          //
          left = ContextState.allocate(1, 0, 0);
          left.writeGlobal(CounterId.fromInt(1), 10L, 20L);
  
          right = ContextState.allocate(1, 0, 0);
          right.writeGlobal(CounterId.fromInt(1), 10L, 30L);
  
          merged = cc.merge(left.context, right.context);
          headerLength = headerSizeLength + headerEltLength;
          assertEquals(headerLength + stepLength, merged.remaining());
          assertEquals(30L, cc.total(merged, ByteBufferAccessor.instance));
          assertEquals(1, merged.getShort(merged.position()));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), merged, headerLength));
          assertEquals(10L, merged.getLong(merged.position() + headerLength + idLength));
          // with equal clock, we should pick the largest value
          assertEquals(30L, merged.getLong(merged.position() + headerLength + idLength + clockLength));
  
          //
          // Test merging global w/ mixed contexts
          //
          left = ContextState.allocate(2, 0, 0);
          left.writeGlobal(CounterId.fromInt(1), 1L, 1L);
          left.writeGlobal(CounterId.fromInt(2), 1L, 1L);
  
          right = ContextState.allocate(0, 1, 1);
          right.writeLocal(CounterId.fromInt(1), 100L, 100L);
          right.writeRemote(CounterId.fromInt(2), 100L, 100L);
  
          // global shards should dominate local/remote, even with lower clock and value
          merged = cc.merge(left.context, right.context);
          headerLength = headerSizeLength + 2 * headerEltLength;
          assertEquals(headerLength + 2 * stepLength, merged.remaining());
          assertEquals(2L, cc.total(merged, ByteBufferAccessor.instance));
          assertEquals(2, merged.getShort(merged.position()));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), merged, headerLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + idLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + idLength + clockLength));
          assertTrue(Util.equalsCounterId(CounterId.fromInt(2), merged, headerLength + stepLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + stepLength + idLength));
          assertEquals(1L, merged.getLong(merged.position() + headerLength + stepLength + idLength + clockLength));
      }
  
      @Test
      public void testTotal()
      {
          ContextState mixed = ContextState.allocate(0, 1, 4);
          mixed.writeRemote(CounterId.fromInt(1), 1L, 1L);
          mixed.writeRemote(CounterId.fromInt(2), 2L, 2L);
          mixed.writeRemote(CounterId.fromInt(4), 4L, 4L);
          mixed.writeRemote(CounterId.fromInt(5), 5L, 5L);
          mixed.writeLocal(CounterId.getLocalId(), 12L, 12L);
          assertEquals(24L, cc.total(mixed.context, ByteBufferAccessor.instance));
  
          ContextState global = ContextState.allocate(3, 0, 0);
          global.writeGlobal(CounterId.fromInt(1), 1L, 1L);
          global.writeGlobal(CounterId.fromInt(2), 2L, 2L);
          global.writeGlobal(CounterId.fromInt(3), 3L, 3L);
          assertEquals(6L, cc.total(global.context, ByteBufferAccessor.instance));
      }
  
      @Test
      public void testClearLocal()
      {
          ContextState state;
          ByteBuffer marked;
          ByteBuffer cleared;
  
          // mark/clear for remote-only contexts is a no-op
          state = ContextState.allocate(0, 0, 1);
          state.writeRemote(CounterId.fromInt(1), 1L, 1L);
  
          assertFalse(cc.shouldClearLocal(state.context, ByteBufferAccessor.instance));
          marked = cc.markLocalToBeCleared(state.context);
          assertEquals(0, marked.getShort(marked.position()));
          assertSame(state.context, marked); // should return the original context
  
          cleared = cc.clearAllLocal(marked, ByteBufferAccessor.instance);
          assertSame(cleared, marked); // shouldn't alter anything either
  
          // a single local shard
          state = ContextState.allocate(0, 1, 0);
          state.writeLocal(CounterId.fromInt(1), 1L, 1L);
  
          assertFalse(cc.shouldClearLocal(state.context, ByteBufferAccessor.instance));
          marked = cc.markLocalToBeCleared(state.context);
          assertTrue(cc.shouldClearLocal(marked, ByteBufferAccessor.instance));
          assertEquals(-1, marked.getShort(marked.position()));
          assertNotSame(state.context, marked); // shouldn't alter in place, as it used to do
  
          cleared = cc.clearAllLocal(marked, ByteBufferAccessor.instance);
          assertFalse(cc.shouldClearLocal(cleared, ByteBufferAccessor.instance));
          assertEquals(0, cleared.getShort(cleared.position()));
  
          // 2 global + 1 local shard
          state = ContextState.allocate(2, 1, 0);
          state.writeLocal(CounterId.fromInt(1), 1L, 1L);
          state.writeGlobal(CounterId.fromInt(2), 2L, 2L);
          state.writeGlobal(CounterId.fromInt(3), 3L, 3L);
  
          assertFalse(cc.shouldClearLocal(state.context, ByteBufferAccessor.instance));
          marked = cc.markLocalToBeCleared(state.context);
          assertTrue(cc.shouldClearLocal(marked, ByteBufferAccessor.instance));
  
          assertEquals(-3, marked.getShort(marked.position()));
          assertEquals(0, marked.getShort(marked.position() + headerSizeLength));
          assertEquals(Short.MIN_VALUE + 1, marked.getShort(marked.position() + headerSizeLength + headerEltLength));
          assertEquals(Short.MIN_VALUE + 2, marked.getShort(marked.position() + headerSizeLength + 2 * headerEltLength));
  
          int headerLength = headerSizeLength + 3 * headerEltLength;
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), marked, headerLength));
          assertEquals(1L, marked.getLong(marked.position() + headerLength + idLength));
          assertEquals(1L, marked.getLong(marked.position() + headerLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(2), marked, headerLength + stepLength));
          assertEquals(2L, marked.getLong(marked.position() + headerLength + stepLength + idLength));
          assertEquals(2L, marked.getLong(marked.position() + headerLength + stepLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(3), marked, headerLength + 2 * stepLength));
          assertEquals(3L, marked.getLong(marked.position() + headerLength + 2 * stepLength + idLength));
          assertEquals(3L, marked.getLong(marked.position() + headerLength + 2 * stepLength + idLength + clockLength));
  
          cleared = cc.clearAllLocal(marked, ByteBufferAccessor.instance);
          assertFalse(cc.shouldClearLocal(cleared, ByteBufferAccessor.instance));
  
          assertEquals(2, cleared.getShort(cleared.position())); // 2 global shards
          assertEquals(Short.MIN_VALUE + 1, cleared.getShort(marked.position() + headerEltLength));
          assertEquals(Short.MIN_VALUE + 2, cleared.getShort(marked.position() + headerSizeLength + headerEltLength));
  
          headerLength = headerSizeLength + 2 * headerEltLength;
          assertTrue(Util.equalsCounterId(CounterId.fromInt(1), cleared, headerLength));
          assertEquals(1L, cleared.getLong(cleared.position() + headerLength + idLength));
          assertEquals(1L, cleared.getLong(cleared.position() + headerLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(2), cleared, headerLength + stepLength));
          assertEquals(2L, cleared.getLong(cleared.position() + headerLength + stepLength + idLength));
          assertEquals(2L, cleared.getLong(cleared.position() + headerLength + stepLength + idLength + clockLength));
  
          assertTrue(Util.equalsCounterId(CounterId.fromInt(3), cleared, headerLength + 2 * stepLength));
          assertEquals(3L, cleared.getLong(cleared.position() + headerLength + 2 * stepLength + idLength));
          assertEquals(3L, cleared.getLong(cleared.position() + headerLength + 2 * stepLength + idLength + clockLength));
  
          // a single global shard - no-op
          state = ContextState.allocate(1, 0, 0);
          state.writeGlobal(CounterId.fromInt(1), 1L, 1L);
  
          assertFalse(cc.shouldClearLocal(state.context, ByteBufferAccessor.instance));
          marked = cc.markLocalToBeCleared(state.context);
          assertEquals(1, marked.getShort(marked.position()));
          assertSame(state.context, marked);
  
          cleared = cc.clearAllLocal(marked, ByteBufferAccessor.instance);
          assertSame(cleared, marked);
      }
  
      @Test
      public void testFindPositionOf()
      {
          ContextState state = ContextState.allocate(3, 3, 3);
  
          state.writeGlobal(CounterId.fromInt(1), 1L, 1L);
          state.writeRemote(CounterId.fromInt(2), 2L, 2L);
          state.writeLocal( CounterId.fromInt(3), 3L, 3L);
          state.writeGlobal(CounterId.fromInt(4), 4L, 4L);
          state.writeRemote(CounterId.fromInt(5), 5L, 5L);
          state.writeLocal( CounterId.fromInt(6), 6L, 6L);
          state.writeGlobal(CounterId.fromInt(7), 7L, 7L);
          state.writeRemote(CounterId.fromInt(8), 8L, 8L);
          state.writeLocal(CounterId.fromInt(9), 9L, 9L);
  
          int headerLength = headerSizeLength + 6 * headerEltLength;
          assertEquals(headerLength, cc.findPositionOf(state.context, CounterId.fromInt(1)));
          assertEquals(headerLength + stepLength, cc.findPositionOf(state.context, CounterId.fromInt(2)));
          assertEquals(headerLength + 2 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(3)));
          assertEquals(headerLength + 3 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(4)));
          assertEquals(headerLength + 4 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(5)));
          assertEquals(headerLength + 5 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(6)));
          assertEquals(headerLength + 6 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(7)));
          assertEquals(headerLength + 7 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(8)));
          assertEquals(headerLength + 8 * stepLength, cc.findPositionOf(state.context, CounterId.fromInt(9)));
  
          assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(0)));
          assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(10)));
          assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(15)));
          assertEquals(-1, cc.findPositionOf(state.context, CounterId.fromInt(20)));
      }
  
      @Test
      public void testGetGlockAndCountOf()
      {
          ContextState state = ContextState.allocate(3, 3, 3);
  
          state.writeGlobal(CounterId.fromInt(1), 1L, 1L);
          state.writeRemote(CounterId.fromInt(2), 2L, 2L);
          state.writeLocal( CounterId.fromInt(3), 3L, 3L);
          state.writeGlobal(CounterId.fromInt(4), 4L, 4L);
          state.writeRemote(CounterId.fromInt(5), 5L, 5L);
          state.writeLocal( CounterId.fromInt(6), 6L, 6L);
          state.writeGlobal(CounterId.fromInt(7), 7L, 7L);
          state.writeRemote(CounterId.fromInt(8), 8L, 8L);
          state.writeLocal(CounterId.fromInt(9), 9L, 9L);
  
          assertEquals(ClockAndCount.create(1L, 1L), cc.getClockAndCountOf(state.context, CounterId.fromInt(1)));
          assertEquals(ClockAndCount.create(2L, 2L), cc.getClockAndCountOf(state.context, CounterId.fromInt(2)));
          assertEquals(ClockAndCount.create(3L, 3L), cc.getClockAndCountOf(state.context, CounterId.fromInt(3)));
          assertEquals(ClockAndCount.create(4L, 4L), cc.getClockAndCountOf(state.context, CounterId.fromInt(4)));
          assertEquals(ClockAndCount.create(5L, 5L), cc.getClockAndCountOf(state.context, CounterId.fromInt(5)));
          assertEquals(ClockAndCount.create(6L, 6L), cc.getClockAndCountOf(state.context, CounterId.fromInt(6)));
          assertEquals(ClockAndCount.create(7L, 7L), cc.getClockAndCountOf(state.context, CounterId.fromInt(7)));
          assertEquals(ClockAndCount.create(8L, 8L), cc.getClockAndCountOf(state.context, CounterId.fromInt(8)));
          assertEquals(ClockAndCount.create(9L, 9L), cc.getClockAndCountOf(state.context, CounterId.fromInt(9)));
  
          assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(0)));
          assertEquals(ClockAndCount.create(0L, 0L), cc.getClockAndCountOf(state.context, CounterId.fromInt(10)));
          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));
      }
  }
  