AST Harrys multi node tests are flakey when multi cell list happens cross instances

patch by David Capwell; reviewed by Caleb Rackliffe for CASSANDRA-20544

test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java

@@ -21,12 +21,19 @@ package org.apache.cassandra.distributed.test.cql3;
 import java.io.IOException;
 
 import accord.utils.RandomSource;
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
 import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.distributed.api.IInvokableInstance;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
+import org.apache.cassandra.utils.Shared;
+import org.apache.cassandra.utils.TimeUUID;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
 
 public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
 {
@@ -53,7 +60,7 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
     @Override
     protected Cluster createCluster() throws IOException
     {
-        return createCluster(mockMultiNode ? 1 : 3, this::clusterConfig);
+        return createCluster(mockMultiNode ? 1 : 3);
     }
 
     @Override
@@ -66,6 +73,12 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
          .set("slow_query_log_timeout", "180s");
     }
 
+    @Override
+    protected void clusterInitializer(ClassLoader cl, int node)
+    {
+        BBHelper.install(cl, node);
+    }
+
     @Override
     protected State createState(RandomSource rs, Cluster cluster)
     {
@@ -131,4 +144,42 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest
             return ConsistencyLevel.NODE_LOCAL;
         }
     }
+
+    /**
+     * This is not a deterministic clock for TimeUUID, but it's a monotonic clock, which means that any instance that gets
+     * a TimeUUID from this clock has the propery that its happens-after all other ones cross all instances.
+     *
+     * This class came around because TimeUUID.Generator.nextUnixMicros works with milliseconds, and when time doesn't
+     * move forward (goes back or test is "too fast") then it becomes an instance local bump-counter; this counter allows
+     * a logically later timeuuid to happens-before a logically earlier one!
+     */
+    @Shared
+    public static class GlobalClock
+    {
+        private static long lastMicros = 0;
+        public synchronized static long nextUnixMicros()
+        {
+            return ++lastMicros;
+        }
+
+        public synchronized static void reset()
+        {
+            // this method isn't actually needed for the property of this class, but it does help isolate any non-deterministic issues
+            lastMicros = 0;
+        }
+    }
+
+    public static class BBHelper
+    {
+        static void install(ClassLoader cl, int nodeNumber)
+        {
+            new ByteBuddy().rebase(TimeUUID.Generator.class)
+                           .method(named("nextUnixMicros"))
+                           .intercept(MethodDelegation.to(GlobalClock.class))
+                           .make()
+                           .load(cl, ClassLoadingStrategy.Default.INJECTION);
+
+            GlobalClock.reset();
+        }
+    }
 }

test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java

@@ -24,6 +24,7 @@ import accord.utils.Property;
 import accord.utils.RandomSource;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 
@@ -46,16 +47,20 @@ public class MultiNodeTokenConflictTest extends SingleNodeTokenConflictTest
         return tbl.unbuild().params(tbl.params.unbuild().readRepair(ReadRepairStrategy.NONE).build()).build();
     }
 
+    @Override
+    protected void clusterConfig(IInstanceConfig c)
+    {
+        c.set("range_request_timeout", "180s")
+         .set("read_request_timeout", "180s")
+         .set("write_request_timeout", "180s")
+         .set("native_transport_timeout", "180s")
+         .set("slow_query_log_timeout", "180s");
+    }
+
     @Override
     protected Cluster createCluster() throws IOException
     {
-        return createCluster(3, c -> {
-            c.set("range_request_timeout", "180s")
-             .set("read_request_timeout", "180s")
-             .set("write_request_timeout", "180s")
-             .set("native_transport_timeout", "180s")
-             .set("slow_query_log_timeout", "180s");
-        });
+        return createCluster(3);
     }
 
     @Override

test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java

@@ -59,7 +59,6 @@ import org.apache.cassandra.db.marshal.InetAddressType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.distributed.Cluster;
-import org.apache.cassandra.distributed.api.IInstanceConfig;
 import org.apache.cassandra.distributed.test.sai.SAIUtil;
 import org.apache.cassandra.harry.model.BytesPartitionState;
 import org.apache.cassandra.schema.ColumnMetadata;
@@ -350,12 +349,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
 
     protected Cluster createCluster() throws IOException
     {
-        return createCluster(1, this::clusterConfig);
-    }
-
-    protected void clusterConfig(IInstanceConfig config)
-    {
-
+        return createCluster(1);
     }
 
     @Test

test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java

@@ -240,7 +240,7 @@ public class SingleNodeTokenConflictTest extends StatefulASTBase
 
     protected Cluster createCluster() throws IOException
     {
-        return createCluster(1, i -> {});
+        return createCluster(1);
     }
 
     @Test

test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java

@@ -27,6 +27,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -137,9 +138,25 @@ public class StatefulASTBase extends TestBaseImpl
         return "ks" + COUNTER.incrementAndGet();
     }
 
-    protected static Cluster createCluster(int nodeCount, Consumer<IInstanceConfig> config) throws IOException
+    protected void clusterConfig(IInstanceConfig config)
+    {
+
+    }
+
+    protected void clusterInitializer(ClassLoader cl, int node)
+    {
+
+    }
+
+    protected Cluster createCluster(int nodeCount) throws IOException
+    {
+        return createCluster(nodeCount, this::clusterConfig, this::clusterInitializer);
+    }
+
+    protected static Cluster createCluster(int nodeCount, Consumer<IInstanceConfig> config, BiConsumer<ClassLoader, Integer> instanceInitializer) throws IOException
     {
         Cluster cluster = Cluster.build(nodeCount)
+                                 .withInstanceInitializer(instanceInitializer)
                                  .withConfig(c -> {
                                      c.with(Feature.NATIVE_PROTOCOL, Feature.NETWORK, Feature.GOSSIP)
                                       // When drop tables or truncate are performed, we attempt to take snapshots.  This can be costly and isn't needed by these tests