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.repair;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.EnumSet;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.LinkedHashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Random;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.function.BiFunction;
  import java.util.function.Consumer;
  import java.util.function.Function;
  import java.util.function.Supplier;
  import javax.annotation.Nullable;
  
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Maps;
  import com.google.common.collect.Sets;
  import org.junit.BeforeClass;
  
  import accord.utils.DefaultRandom;
  import accord.utils.Gen;
  import accord.utils.Gens;
  import accord.utils.RandomSource;
  import org.agrona.collections.LongHashSet;
  import org.apache.cassandra.ServerTestUtils;
  import org.apache.cassandra.concurrent.ExecutorBuilder;
  import org.apache.cassandra.concurrent.ExecutorBuilderFactory;
  import org.apache.cassandra.concurrent.ExecutorFactory;
  import org.apache.cassandra.concurrent.ExecutorPlus;
  import org.apache.cassandra.concurrent.InfiniteLoopExecutor;
  import org.apache.cassandra.concurrent.Interruptible;
  import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
  import org.apache.cassandra.concurrent.SequentialExecutorPlus;
  import org.apache.cassandra.concurrent.SimulatedExecutorFactory;
  import org.apache.cassandra.concurrent.Stage;
  import org.apache.cassandra.config.Config;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.config.UnitConfigOverride;
  import org.apache.cassandra.cql3.CQLTester;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.Digest;
  import org.apache.cassandra.db.SystemKeyspace;
  import org.apache.cassandra.db.compaction.ICompactionManager;
  import org.apache.cassandra.db.marshal.EmptyType;
  import org.apache.cassandra.db.repair.CassandraTableRepairManager;
  import org.apache.cassandra.db.repair.PendingAntiCompaction;
  import org.apache.cassandra.dht.Murmur3Partitioner;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
  import org.apache.cassandra.gms.ApplicationState;
  import org.apache.cassandra.gms.EndpointState;
  import org.apache.cassandra.gms.HeartBeatState;
  import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
  import org.apache.cassandra.gms.IFailureDetector;
  import org.apache.cassandra.gms.IGossiper;
  import org.apache.cassandra.gms.VersionedValue;
  import org.apache.cassandra.io.util.DataInputBuffer;
  import org.apache.cassandra.io.util.DataOutputBuffer;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.locator.LocalStrategy;
  import org.apache.cassandra.locator.Locator;
  import org.apache.cassandra.locator.RangesAtEndpoint;
  import org.apache.cassandra.net.IVerbHandler;
  import org.apache.cassandra.net.Message;
  import org.apache.cassandra.net.MessageDelivery;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.net.SimulatedMessageDelivery;
  import org.apache.cassandra.net.SimulatedMessageDelivery.SimulatedMessageReceiver;
  import org.apache.cassandra.repair.RepairGenerators.PreviewType;
  import org.apache.cassandra.repair.RepairGenerators.RepairType;
  import org.apache.cassandra.repair.messages.RepairMessage;
  import org.apache.cassandra.repair.messages.RepairOption;
  import org.apache.cassandra.repair.messages.ValidationResponse;
  import org.apache.cassandra.repair.state.Completable;
  import org.apache.cassandra.repair.state.CoordinatorState;
  import org.apache.cassandra.repair.state.JobState;
  import org.apache.cassandra.repair.state.ParticipateState;
  import org.apache.cassandra.repair.state.SessionState;
  import org.apache.cassandra.repair.state.ValidationState;
  import org.apache.cassandra.schema.KeyspaceMetadata;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.schema.SchemaConstants;
  import org.apache.cassandra.schema.SystemDistributedKeyspace;
  import org.apache.cassandra.schema.TableId;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.schema.Tables;
  import org.apache.cassandra.service.ActiveRepairService;
  import org.apache.cassandra.service.StorageService;
  import org.apache.cassandra.service.accord.AccordService;
  import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupComplete;
  import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupHistory;
  import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupRequest;
  import org.apache.cassandra.service.paxos.cleanup.PaxosCleanupResponse;
  import org.apache.cassandra.service.paxos.cleanup.PaxosFinishPrepareCleanup;
  import org.apache.cassandra.service.paxos.cleanup.PaxosRepairState;
  import org.apache.cassandra.service.paxos.cleanup.PaxosStartPrepareCleanup;
  import org.apache.cassandra.streaming.StreamEventHandler;
  import org.apache.cassandra.streaming.StreamReceiveException;
  import org.apache.cassandra.streaming.StreamSession;
  import org.apache.cassandra.streaming.StreamState;
  import org.apache.cassandra.streaming.StreamingChannel;
  import org.apache.cassandra.streaming.StreamingDataInputPlus;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.tcm.membership.Location;
  import org.apache.cassandra.tools.nodetool.Repair;
  import org.apache.cassandra.utils.AbstractTypeGenerators;
  import org.apache.cassandra.utils.CassandraGenerators;
  import org.apache.cassandra.utils.Clock;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.FailingBiConsumer;
  import org.apache.cassandra.utils.Generators;
  import org.apache.cassandra.utils.MBeanWrapper;
  import org.apache.cassandra.utils.MerkleTree;
  import org.apache.cassandra.utils.MerkleTrees;
  import org.apache.cassandra.utils.NoSpamLogger;
  import org.apache.cassandra.utils.TimeUUID;
  import org.apache.cassandra.utils.concurrent.ImmediateFuture;
  import org.apache.cassandra.utils.progress.ProgressEventType;
  import org.assertj.core.api.Assertions;
  import org.mockito.Mockito;
  import org.quicktheories.impl.JavaRandom;
  import picocli.CommandLine;
  
  import static org.apache.cassandra.config.CassandraRelevantProperties.ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL;
  import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOBAL;
  import static org.apache.cassandra.config.CassandraRelevantProperties.ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION;
  
  public abstract class FuzzTestBase extends CQLTester.InMemory
  {
      private static final int MISMATCH_NUM_PARTITIONS = 1;
      private static final Gen<String> IDENTIFIER_GEN = Generators.toGen(Generators.IDENTIFIER_GEN);
      private static final Gen<String> KEYSPACE_NAME_GEN = Generators.toGen(CassandraGenerators.KEYSPACE_NAME_GEN);
      private static final Gen<TableId> TABLE_ID_GEN = Generators.toGen(CassandraGenerators.TABLE_ID_GEN);
      private static final Gen<InetAddressAndPort> ADDRESS_W_PORT = Generators.toGen(CassandraGenerators.INET_ADDRESS_AND_PORT_GEN);
  
      private static boolean SETUP_SCHEMA = false;
      static String KEYSPACE;
      static List<String> TABLES;
  
      @BeforeClass
      public static void setUpClass()
      {
          ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(true);
          ACCORD_REPAIR_RANGE_STEP_UPDATE_INTERVAL.setInt(1);
          CLOCK_GLOBAL.setString(ClockAccess.class.getName());
          // when running in CI an external actor will replace the test configs based off the test type (such as trie, cdc, etc.), this could then have failing tests
          // that do not repo with the same seed!  To fix that, go to UnitConfigOverride and update the config type to match the one that failed in CI, this should then
          // use the same config, so the seed should not reproduce.
          UnitConfigOverride.maybeOverrideConfig();
  
          DatabaseDescriptor.daemonInitialization();
  
          // in-memory fs (jimfs) we use for testing purposes does not handle direct mode well,
          // it is evalauted to direct only in case we are running -latest test profile where
          // commitlog_disk_access_mode is set to auto which would evaluate it to "direct" which fails,
          // other test profiles (normal or compression) would run with in-memory fs just fine,
          // it is just "direct" which can not be run with in-memory fs, in that case,
          // we set it forcibly to mmap to bypass that
          // please keep in mind this is a static field, impacting every test running in the same jvm
          if (DatabaseDescriptor.getCommitLogWriteDiskAccessMode() == Config.DiskAccessMode.direct)
              DatabaseDescriptor.setCommitLogWriteDiskAccessMode(Config.DiskAccessMode.mmap);
  
          DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance); // TOOD (coverage): random select
          StreamingChannel.Factory.Global.unsafeSet(new StreamingChannel.Factory()
          {
              private final AtomicInteger counter = new AtomicInteger();
  
              @Override
              public StreamingChannel create(InetSocketAddress to, int messagingVersion, StreamingChannel.Kind kind) throws IOException
              {
                  StreamingChannel mock = Mockito.mock(StreamingChannel.class);
                  int id = counter.incrementAndGet();
                  StreamSession session = Mockito.mock(StreamSession.class);
                  StreamReceiveException access = new StreamReceiveException(session, "mock access rejected");
                  StreamingDataInputPlus input = Mockito.mock(StreamingDataInputPlus.class, invocationOnMock -> {
                      throw access;
                  });
                  Mockito.doNothing().when(input).close();
                  Mockito.when(mock.in()).thenReturn(input);
                  Mockito.when(mock.id()).thenReturn(id);
                  Mockito.when(mock.peer()).thenReturn(to);
                  Mockito.when(mock.connectedTo()).thenReturn(to);
                  Mockito.when(mock.send(Mockito.any())).thenReturn(ImmediateFuture.success(null));
                  Mockito.when(mock.close()).thenReturn(ImmediateFuture.success(null));
                  return mock;
              }
          });
          ExecutorFactory delegate = ExecutorFactory.Global.executorFactory();
          ExecutorFactory.Global.unsafeSet(new ExecutorFactory()
          {
              @Override
              public LocalAwareSubFactory localAware()
              {
                  return delegate.localAware();
              }
  
              @Override
              public ScheduledExecutorPlus scheduled(boolean executeOnShutdown, String name, int priority, SimulatorSemantics simulatorSemantics)
              {
                  return delegate.scheduled(executeOnShutdown, name, priority, simulatorSemantics);
              }
  
              private boolean shouldMock()
              {
                  return StackWalker.getInstance().walk(frame -> {
                      StackWalker.StackFrame caller = frame.skip(3).findFirst().get();
                      return caller.getClassName().startsWith("org.apache.cassandra.streaming.");
                  });
              }
  
              @Override
              public Thread startThread(String name, Runnable runnable, SystemThreadTag systemTag, SimulatorThreadTag simulatorTag)
              {
                  if (shouldMock()) return new Thread();
                  return delegate.startThread(name, runnable, systemTag, simulatorTag);
              }
  
              @Override
              public Interruptible infiniteLoop(String name, Interruptible.Task task, InfiniteLoopExecutor.SimulatorSafe simulatorSafe, SystemThreadTag systemTag, InfiniteLoopExecutor.Interrupts interrupts)
              {
                  return delegate.infiniteLoop(name, task, simulatorSafe, systemTag, interrupts);
              }
  
              @Override
              public ThreadGroup newThreadGroup(String name)
              {
                  return delegate.newThreadGroup(name);
              }
  
              @Override
              public ExecutorBuilderFactory<ExecutorPlus, SequentialExecutorPlus> withJmx(String jmxPath)
              {
                  return delegate.withJmx(jmxPath);
              }
  
              @Override
              public ExecutorBuilder<? extends SequentialExecutorPlus> configureSequential(String name)
              {
                  return delegate.configureSequential(name);
              }
  
              @Override
              public ExecutorBuilder<? extends ExecutorPlus> configurePooled(String name, int threads)
              {
                  return delegate.configurePooled(name, threads);
              }
          });
  
          // will both make sure this is loaded and used
          if (!(Clock.Global.clock() instanceof ClockAccess)) throw new IllegalStateException("Unable to override clock");
  
          // set the repair rcp timeout high so we don't hit it... this class is mostly testing repair reaching success
          // so don't want to deal with unlucky histories...
          DatabaseDescriptor.setRepairRpcTimeout(TimeUnit.DAYS.toMillis(1));
  
          // make sure accord is enabled as accord has custom repair steps
          DatabaseDescriptor.setAccordTransactionsEnabled(true);
  
          InMemory.setUpClass();
  
          MessagingService.instance().listen();
      }
  
      public static void setupSchema()
      {
          // StorageService can not be mocked out, nor can ColumnFamilyStores, so make sure that the keyspace is a "local" keyspace to avoid replication as the peers don't actually exist for replication
          schemaChange(String.format("CREATE KEYSPACE %s WITH REPLICATION = {'class': '%s'}", SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, HackStrat.class.getName()));
          for (TableMetadata table : SystemDistributedKeyspace.metadata().tables)
              schemaChange(table.toCqlString(true, false, false));
  
          createSchema();
      }
  
      protected void cleanupRepairTables()
      {
          for (String table : Arrays.asList(SystemKeyspace.REPAIRS))
              execute(String.format("TRUNCATE %s.%s", SchemaConstants.SYSTEM_KEYSPACE_NAME, table));
      }
  
      private static void createSchema()
      {
          // The main reason to use random here with a fixed seed is just to have a set of tables that are not hard coded.
          // The tables will have diversity to them that most likely doesn't matter to repair (hence why the tables are shared), but
          // is useful just in case some assumptions change.
          RandomSource rs = new DefaultRandom(42);
          String ks = KEYSPACE_NAME_GEN.next(rs);
          List<String> tableNames = Gens.lists(IDENTIFIER_GEN).unique().ofSizeBetween(10, 100).next(rs);
          JavaRandom qt = new JavaRandom(rs.asJdkRandom());
          Tables.Builder tableBuilder = Tables.builder();
          List<TableId> ids = Gens.lists(TABLE_ID_GEN).unique().ofSize(tableNames.size()).next(rs);
          for (int i = 0; i < tableNames.size(); i++)
          {
              String name = tableNames.get(i);
              TableId id = ids.get(i);
              TableMetadata tableMetadata = new CassandraGenerators.TableMetadataBuilder().withKeyspaceName(ks).withTableName(name).withTableId(id).withTableKinds(TableMetadata.Kind.REGULAR)
                                                                                          // shouldn't matter, just wanted to avoid UDT as that needs more setup
                                                                                          .withDefaultTypeGen(AbstractTypeGenerators.builder().withTypeKinds(AbstractTypeGenerators.TypeKind.PRIMITIVE).withoutPrimitive(EmptyType.instance)).build().generate(qt);
              tableBuilder.add(tableMetadata);
          }
          KeyspaceParams params = KeyspaceParams.simple(3);
          KeyspaceMetadata metadata = KeyspaceMetadata.create(ks, params, tableBuilder.build());
  
          // create
          schemaChange(metadata.toCqlString(true, false, false));
          KEYSPACE = ks;
          for (TableMetadata table : metadata.tables)
              schemaChange(table.toCqlString(true, false, false));
          TABLES = tableNames;
      }
  
      static void enableMessageFaults(Cluster cluster)
      {
          cluster.allowedMessageFaults(new BiFunction<>()
          {
              private final LongHashSet noFaults = new LongHashSet();
              private final LongHashSet allowDrop = new LongHashSet();
  
              @Override
              public Set<Faults> apply(Cluster.Node node, Message<?> message)
              {
                  if (RepairMessage.ALLOWS_RETRY.contains(message.verb()))
                  {
                      allowDrop.add(message.id());
                      return Faults.DROPPED;
                  }
                  switch (message.verb())
                  {
                      // these messages are not resilent to ephemeral issues
                      case STATUS_REQ:
                      case STATUS_RSP:
                      // paxos repair does not support faults and will cause a TIMEOUT error, failing the repair
                      case PAXOS2_CLEANUP_COMPLETE_REQ:
                      case PAXOS2_CLEANUP_REQ:
                      case PAXOS2_CLEANUP_RSP2:
                      case PAXOS2_CLEANUP_START_PREPARE_REQ:
                      case PAXOS2_CLEANUP_FINISH_PREPARE_REQ:
                          noFaults.add(message.id());
                          return Faults.NONE;
                      default:
                          if (noFaults.contains(message.id())) return Faults.NONE;
                          if (allowDrop.contains(message.id())) return Faults.DROPPED;
                          // was a new message added and the test not updated?
                          IllegalStateException e = new IllegalStateException("Verb: " + message.verb());
                          cluster.failures.add(e);
                          throw e;
                  }
              }
          });
      }
  
      static void runAndAssertSuccess(Cluster cluster, int example, boolean shouldSync, RepairCoordinator repair)
      {
          cluster.processAll();
          assertSuccess(cluster, example, shouldSync, repair);
      }
  
      static void assertSuccess(Cluster cluster, int example, boolean shouldSync, RepairCoordinator repair)
      {
          Completable.Result result = repair.state.getResult();
          Assertions.assertThat(result)
                    .describedAs("Expected repair to have completed with success, but is still running... %s; example %d", repair.state, example).isNotNull()
                    .describedAs("Unexpected state: %s -> %s; example %d", repair.state, result, example).isEqualTo(Completable.Result.success(repairSuccessMessage(repair)));
          Assertions.assertThat(repair.state.getStateTimesMillis().keySet()).isEqualTo(EnumSet.allOf(CoordinatorState.State.class));
          Assertions.assertThat(repair.state.getSessions()).isNotEmpty();
          boolean shouldSnapshot = repair.state.options.getParallelism() != RepairParallelism.PARALLEL
                                   && (!repair.state.options.isIncremental() || repair.state.options.isPreview());
          for (SessionState session : repair.state.getSessions())
          {
              Assertions.assertThat(session.getStateTimesMillis().keySet()).isEqualTo(EnumSet.allOf(SessionState.State.class));
              Assertions.assertThat(session.getJobs()).isNotEmpty();
              for (JobState job : session.getJobs())
              {
                  EnumSet<JobState.State> expected = EnumSet.allOf(JobState.State.class);
                  if (!shouldSnapshot)
                  {
                      expected.remove(JobState.State.SNAPSHOT_START);
                      expected.remove(JobState.State.SNAPSHOT_COMPLETE);
                  }
                  if (!shouldSync)
                  {
                      expected.remove(JobState.State.STREAM_START);
                  }
                  Set<JobState.State> actual = job.getStateTimesMillis().keySet();
                  Assertions.assertThat(actual).isEqualTo(expected);
              }
          }
  
          assertParticipateResult(cluster, repair, Completable.Result.Kind.SUCCESS);
      }
  
      protected static void assertParticipateResult(Cluster cluster, RepairCoordinator repair, Completable.Result.Kind kind)
      {
          for (InetAddressAndPort participate : Sets.union(Collections.singleton(repair.ctx.broadcastAddressAndPort()), repair.state.getNeighborsAndRanges().participants))
          {
              assertParticipateResult(cluster, participate, repair.state.id, kind);
          }
      }
  
      protected static void assertParticipateResult(Cluster cluster, InetAddressAndPort participate, TimeUUID id, Completable.Result.Kind kind)
      {
          Cluster.Node node = cluster.nodes.get(participate);
          ParticipateState state = node.repair().participate(id);
          Assertions.assertThat(state).describedAs("Node %s is missing ParticipateState", node).isNotNull();
          Completable.Result particpateResult = state.getResult();
          Assertions.assertThat(particpateResult).describedAs("Node %s has the ParticipateState as still pending", node).isNotNull();
          Assertions.assertThat(particpateResult.kind).isEqualTo(kind);
      }
  
      static String repairSuccessMessage(RepairCoordinator repair)
      {
          RepairOption options = repair.state.options;
          if (options.isPreview())
          {
              String suffix;
              switch (options.getPreviewKind())
              {
                  case UNREPAIRED:
                  case ALL:
                      suffix = "Previewed data was in sync";
                      break;
                  case REPAIRED:
                      suffix = "Repaired data is in sync";
                      break;
                  default:
                      throw new IllegalArgumentException("Unexpected preview repair kind: " + options.getPreviewKind());
              }
              return "Repair preview completed successfully; " + suffix;
          }
          return "Repair completed successfully";
      }
  
      InetAddressAndPort pickParticipant(RandomSource rs, Cluster.Node coordinator, RepairCoordinator repair)
      {
          if (repair.state.isComplete())
              throw new IllegalStateException("Repair is completed! " + repair.state.getResult());
          List<InetAddressAndPort> participants = new ArrayList<>(repair.state.getNeighborsAndRanges().participants.size() + 1);
          if (rs.nextBoolean()) participants.add(coordinator.broadcastAddressAndPort());
          participants.addAll(repair.state.getNeighborsAndRanges().participants);
          participants.sort(Comparator.naturalOrder());
  
          return participants.get(rs.nextInt(participants.size()));
      }
  
      static void addMismatch(RandomSource rs, ColumnFamilyStore cfs, Validator validator)
      {
          ValidationState state = validator.state;
          int maxDepth = DatabaseDescriptor.getRepairSessionMaxTreeDepth();
          state.phase.start(MISMATCH_NUM_PARTITIONS, 1024);
  
          MerkleTrees trees = new MerkleTrees(cfs.getPartitioner());
          for (Range<Token> range : validator.desc.ranges)
          {
              int depth = (int) Math.min(Math.ceil(Math.log(MISMATCH_NUM_PARTITIONS) / Math.log(2)), maxDepth);
              trees.addMerkleTree((int) Math.pow(2, depth), range);
          }
          Set<Token> allTokens = new HashSet<>();
          for (Range<Token> range : validator.desc.ranges)
          {
              Gen<Token> gen = Generators.toGen(CassandraGenerators.tokensInRange(range));
              Set<Token> tokens = new LinkedHashSet<>();
              for (int i = 0, size = rs.nextInt(1, 10); i < size; i++)
              {
                  for (int attempt = 0; !tokens.add(gen.next(rs)) && attempt < 5; attempt++)
                  {
                  }
              }
              // tokens may or may not be of the expected size; this depends on how wide the range is
              for (Token token : tokens)
                  trees.split(token);
              allTokens.addAll(tokens);
          }
          for (Token token : allTokens)
          {
              findCorrectRange(trees, token, range -> {
                  Digest digest = Digest.forValidator();
                  digest.update(ByteBuffer.wrap(token.toString().getBytes(StandardCharsets.UTF_8)));
                  range.addHash(new MerkleTree.RowHash(token, digest.digest(), 1));
              });
          }
          state.partitionsProcessed++;
          state.bytesRead = 1024;
          state.phase.sendingTrees();
          Stage.ANTI_ENTROPY.execute(() -> {
              state.phase.success();
              validator.respond(new ValidationResponse(validator.desc, trees));
          });
      }
  
      private static void findCorrectRange(MerkleTrees trees, Token token, Consumer<MerkleTree.TreeRange> fn)
      {
          MerkleTrees.TreeRangeIterator it = trees.rangeIterator();
          while (it.hasNext())
          {
              MerkleTree.TreeRange next = it.next();
              if (next.contains(token))
              {
                  fn.accept(next);
                  return;
              }
          }
      }
  
      static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, List<String> tableNames)
      {
          return repairOption(rs, coordinator, ks, Gens.lists(Gens.pick(tableNames)).ofSizeBetween(1, tableNames.size()), Gens.enums().all(RepairType.class), Gens.enums().all(PreviewType.class), Gens.enums().all(RepairParallelism.class));
      }
  
      static RepairOption irOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen<List<String>> tablesGen)
      {
          return repairOption(rs, coordinator, ks, tablesGen, Gens.constant(RepairType.IR), Gens.constant(PreviewType.NONE), Gens.enums().all(RepairParallelism.class));
      }
  
      static RepairOption previewOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen<List<String>> tablesGen)
      {
          return repairOption(rs, coordinator, ks, tablesGen, Gens.constant(RepairType.FULL), Gens.constant(PreviewType.REPAIRED), Gens.enums().all(RepairParallelism.class));
      }
  
      private static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen<List<String>> tablesGen, Gen<RepairType> repairTypeGen, Gen<PreviewType> previewTypeGen, Gen<RepairParallelism> repairParallelismGen)
      {
          Repair repair = new Repair();
          List<String> args = new RepairGenerators.Builder(tablesGen.map(l -> ImmutableList.<String>builderWithExpectedSize(l.size() + 1).add(ks).addAll(l).build()))
                                  .withType(repairTypeGen)
                                  .withPreviewType(previewTypeGen)
                                  .withParallelism(repairParallelismGen)
                                  .withRanges(i -> RepairGenerators.PRIMARY_RANGE)
                                  .build()
                                  .next(rs);
          new CommandLine(repair).parseArgs(args.toArray(new String[0]));
          RepairOption options = RepairOption.parse(repair.createOptions(() -> "test", repair.tables), DatabaseDescriptor.getPartitioner());
  
          if (options.getRanges().isEmpty())
          {
              if (options.isPrimaryRange())
              {
                  // when repairing only primary range, neither dataCenters nor hosts can be set
                  if (options.getDataCenters().isEmpty() && options.getHosts().isEmpty())
                      options.getRanges().addAll(coordinator.getPrimaryRanges(ks));
                      // except dataCenters only contain local DC (i.e. -local)
                  else if (options.isInLocalDCOnly())
                      options.getRanges().addAll(coordinator.getPrimaryRangesWithinDC(ks));
                  else
                      throw new IllegalArgumentException("You need to run primary range repair on all nodes in the cluster.");
              }
              else
              {
                  Iterables.addAll(options.getRanges(), coordinator.getLocalReplicas(ks).onlyFull().ranges());
              }
          }
          return options;
      }
  
      enum Faults
      {
          DELAY, DROP;
  
          public static final Set<Faults> NONE = Collections.emptySet();
          public static final Set<Faults> DROPPED = EnumSet.of(DELAY, DROP);
      }
  
      private static class Connection
      {
          final InetAddressAndPort from, to;
  
          private Connection(InetAddressAndPort from, InetAddressAndPort to)
          {
              this.from = from;
              this.to = to;
          }
  
          @Override
          public boolean equals(Object o)
          {
              if (this == o) return true;
              if (o == null || getClass() != o.getClass()) return false;
              Connection that = (Connection) o;
              return from.equals(that.from) && to.equals(that.to);
          }
  
          @Override
          public int hashCode()
          {
              return Objects.hash(from, to);
          }
  
          @Override
          public String toString()
          {
              return "Connection{" + "from=" + from + ", to=" + to + '}';
          }
      }
  
      interface MessageListener
      {
          default void preHandle(Cluster.Node node, Message<?> msg) {}
      }
  
      static class Cluster
      {
          private static final FailingBiConsumer<ColumnFamilyStore, Validator> DEFAULT_VALIDATION = ValidationManager::doValidation;
  
          final Map<InetAddressAndPort, Node> nodes;
          private final IFailureDetector failureDetector = Mockito.mock(IFailureDetector.class);
          private final Locator locator = Mockito.mock(Locator.class);
          private final SimulatedExecutorFactory globalExecutor;
          final ScheduledExecutorPlus unorderedScheduled;
          final ExecutorPlus orderedExecutor;
          private final Gossip gossiper = new Gossip();
          private final MBeanWrapper mbean = Mockito.mock(MBeanWrapper.class);
          private final List<Throwable> failures = new ArrayList<>();
          private final List<MessageListener> listeners = new ArrayList<>();
          private final RandomSource rs;
          private BiFunction<Node, Message<?>, Set<Faults>> allowedMessageFaults = (a, b) -> Collections.emptySet();
          private final Map<Connection, Supplier<Boolean>> networkDrops = new HashMap<>();
  
          Cluster(RandomSource rs)
          {
              ClockAccess.includeThreadAsOwner();
              this.rs = rs;
              globalExecutor = new SimulatedExecutorFactory(rs);
              orderedExecutor = globalExecutor.configureSequential("ignore").build();
              unorderedScheduled = globalExecutor.scheduled("ignored");
  
  
  
              // We run tests in an isolated JVM per class, so not cleaing up is safe... but if that assumption ever changes, will need to cleanup
              Stage.ANTI_ENTROPY.unsafeSetExecutor(orderedExecutor);
              Stage.MISC.unsafeSetExecutor(orderedExecutor);
              Stage.INTERNAL_RESPONSE.unsafeSetExecutor(unorderedScheduled);
              Mockito.when(failureDetector.isAlive(Mockito.any())).thenReturn(true);
              Thread expectedThread = Thread.currentThread();
              NoSpamLogger.unsafeSetClock(() -> {
                  if (Thread.currentThread() != expectedThread)
                      throw new AssertionError("NoSpamLogger.Clock accessed outside of fuzzing...");
                  return globalExecutor.nanoTime();
              });
  
              int numNodes = rs.nextInt(3, 10);
              List<String> dcs = Gens.lists(IDENTIFIER_GEN).unique().ofSizeBetween(1, Math.min(10, numNodes)).next(rs);
              Map<InetAddressAndPort, Node> nodes = Maps.newHashMapWithExpectedSize(numNodes);
              Gen<Token> tokenGen = Generators.toGen(CassandraGenerators.token(DatabaseDescriptor.getPartitioner()));
              Gen<UUID> hostIdGen = Generators.toGen(Generators.UUID_RANDOM_GEN);
              Set<Token> tokens = new HashSet<>();
              Set<UUID> hostIds = new HashSet<>();
              for (int i = 0; i < numNodes; i++)
              {
                  InetAddressAndPort addressAndPort = ADDRESS_W_PORT.next(rs);
                  while (nodes.containsKey(addressAndPort)) addressAndPort = ADDRESS_W_PORT.next(rs);
                  Token token;
                  while (!tokens.add(token = tokenGen.next(rs)))
                  {
                  }
                  UUID hostId;
                  while (!hostIds.add(hostId = hostIdGen.next(rs)))
                  {
                  }
  
                  String dc = rs.pick(dcs);
                  String rack = "rack";
                  Location location = new Location(dc, rack);
                  Mockito.when(locator.location(Mockito.eq(addressAndPort))).thenReturn(location);
  
                  VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(DatabaseDescriptor.getPartitioner());
                  EndpointState state = new EndpointState(new HeartBeatState(42, 42));
                  state.addApplicationState(ApplicationState.STATUS, valueFactory.normal(Collections.singleton(token)));
                  state.addApplicationState(ApplicationState.STATUS_WITH_PORT, valueFactory.normal(Collections.singleton(token)));
                  state.addApplicationState(ApplicationState.HOST_ID, valueFactory.hostId(hostId));
                  state.addApplicationState(ApplicationState.TOKENS, valueFactory.tokens(Collections.singleton(token)));
                  state.addApplicationState(ApplicationState.DC, valueFactory.datacenter(dc));
                  state.addApplicationState(ApplicationState.RACK, valueFactory.rack(rack));
                  state.addApplicationState(ApplicationState.RELEASE_VERSION, valueFactory.releaseVersion());
  
                  gossiper.endpoints.put(addressAndPort, state);
  
                  Node node = new Node(hostId, addressAndPort, Collections.singletonList(token), new Messaging(addressAndPort));
                  nodes.put(addressAndPort, node);
              }
              this.nodes = nodes;
              ServerTestUtils.recreateCMS();
              assert ClusterMetadata.current().directory.isEmpty() : ClusterMetadata.current().directory;
              for (Node inst : nodes.values())
              {
                  ClusterMetadataTestHelper.register(inst.broadcastAddressAndPort());
                  ClusterMetadataTestHelper.join(inst.broadcastAddressAndPort(), inst.tokens());
              }
              List<InetAddressAndPort> addresses = new ArrayList<>(nodes.keySet());
              addresses.sort(Comparator.naturalOrder());
              AccordService.unsafeSetNoop();
  
              setupSchema();
          }
  
          public Closeable addListener(MessageListener listener)
          {
              listeners.add(listener);
              return () -> removeListener(listener);
          }
  
          public void removeListener(MessageListener listener)
          {
              listeners.remove(listener);
          }
  
          public void allowedMessageFaults(BiFunction<Node, Message<?>, Set<Faults>> fn)
          {
              this.allowedMessageFaults = fn;
          }
  
          public void checkFailures()
          {
              if (Thread.interrupted())
                  failures.add(new InterruptedException());
              if (failures.isEmpty()) return;
              AssertionError error = new AssertionError("Unexpected exceptions found");
              failures.forEach(error::addSuppressed);
              failures.clear();
              throw error;
          }
  
          public boolean processOne()
          {
              boolean result = globalExecutor.processOne();
              checkFailures();
              return result;
          }
  
          public void processAll()
          {
              while (processOne())
              {
              }
          }
  
          private SimulatedMessageDelivery.Action action(InetAddressAndPort self, Message<?> msg, InetAddressAndPort to)
          {
              boolean toSelf = self.equals(to);
              Node node = nodes.get(to);
              Set<Faults> allowedFaults = allowedMessageFaults.apply(node, msg);
              if (allowedFaults.contains(Faults.DROP) && !toSelf && networkDrops(self, to)) return SimulatedMessageDelivery.Action.DROP_PARTITIONED;
              return SimulatedMessageDelivery.Action.DELIVER;
          }
  
          private boolean networkDrops(InetAddressAndPort self, InetAddressAndPort to)
          {
              return networkDrops.computeIfAbsent(new Connection(self, to), ignore -> Gens.bools().biasedRepeatingRuns(rs.nextInt(1, 11) / 100.0D, rs.nextInt(3, 15)).asSupplier(rs)).get();
          }
  
          private class Messaging extends SimulatedMessageDelivery
          {
              private Messaging(InetAddressAndPort broadcastAddressAndPort)
              {
                  super(broadcastAddressAndPort,
                        Cluster.this::action,
                        new NetworkDelaySupplier()
                        {
                            private final NetworkDelaySupplier delegate = SimulatedMessageDelivery.randomDelay(rs);
                            @Nullable
                            @Override
                            public Duration jitter(Message<?> msg, InetAddressAndPort to)
                            {
                                Set<Faults> allowedFaults = allowedMessageFaults.apply(nodes.get(to), msg);
                                if (!allowedFaults.contains(Faults.DELAY) || broadcastAddressAndPort.equals(to))
                                    return null;
                                return delegate.jitter(msg, to);
                            }
                        },
                        (to, msg) -> unorderedScheduled.submit(() -> nodes.get(to).handle(msg)),
                        (action, to, msg) -> logger.warn("{} message {}", action, msg),
                        unorderedScheduled::schedule,
                        failures::add);
              }
          }
  
          private class Gossip implements IGossiper
          {
              private final Map<InetAddressAndPort, EndpointState> endpoints = new HashMap<>();
  
              @Override
              public void register(IEndpointStateChangeSubscriber subscriber)
              {
  
              }
  
              @Override
              public void unregister(IEndpointStateChangeSubscriber subscriber)
              {
  
              }
  
              @Nullable
              @Override
              public EndpointState getEndpointStateForEndpoint(InetAddressAndPort ep)
              {
                  return endpoints.get(ep);
              }
  
              @Override
              public void notifyFailureDetector(Map<InetAddressAndPort, EndpointState> remoteEpStateMap)
              {
  
              }
  
              @Override
              public void applyStateLocally(Map<InetAddressAndPort, EndpointState> epStateMap)
              {
                  // If we were testing paxos this would be wrong...
                  // CASSANDRA-18917 added support for simulating Gossip, but gossip issues were found so couldn't merge that patch...
                  // For the paxos repair, since we don't care about paxos messages, this is ok to no-op for now, but if paxos cleanup
                  // ever was to be tested this logic would need to be implemented
              }
          }
  
          class Node implements SharedContext
          {
              private final ICompactionManager compactionManager = Mockito.mock(ICompactionManager.class);
              final UUID hostId;
              final InetAddressAndPort addressAndPort;
              final Collection<Token> tokens;
              final ActiveRepairService activeRepairService;
              final SimulatedMessageReceiver receiver;
              final Messaging messaging;
              final IValidationManager validationManager;
              private FailingBiConsumer<ColumnFamilyStore, Validator> doValidation = DEFAULT_VALIDATION;
              final PaxosRepairState paxosRepairState;
              private final StreamExecutor defaultStreamExecutor = plan -> {
                  long delayNanos = rs.nextLong(TimeUnit.SECONDS.toNanos(5), TimeUnit.MINUTES.toNanos(10));
                  unorderedScheduled.schedule(() -> {
                      StreamState success = new StreamState(plan.planId(), plan.streamOperation(), Collections.emptySet());
                      for (StreamEventHandler handler : plan.handlers())
                          handler.onSuccess(success);
                  }, delayNanos, TimeUnit.NANOSECONDS);
                  return null;
              };
              private StreamExecutor streamExecutor = defaultStreamExecutor;
  
              private Node(UUID hostId, InetAddressAndPort addressAndPort, Collection<Token> tokens, Messaging messaging)
              {
                  this.hostId = hostId;
                  this.addressAndPort = addressAndPort;
                  this.tokens = tokens;
                  this.messaging = messaging;
                  this.activeRepairService = new ActiveRepairService(this);
                  this.paxosRepairState = new PaxosRepairState(this);
                  this.validationManager = (cfs, validator) -> unorderedScheduled.submit(() -> {
                      try
                      {
                          doValidation.acceptOrFail(cfs, validator);
                      }
                      catch (Throwable e)
                      {
                          validator.fail(e);
                      }
                  });
                  this.receiver = messaging.receiver(new IVerbHandler<>()
                  {
                      private final RepairMessageVerbHandler repairVerbHandler = new RepairMessageVerbHandler(Node.this);
                      private final IVerbHandler<PaxosStartPrepareCleanup.Request> paxosStartPrepareCleanup = PaxosStartPrepareCleanup.createVerbHandler(Node.this);
                      private final IVerbHandler<PaxosCleanupRequest> paxosCleanupRequestIVerbHandler = PaxosCleanupRequest.createVerbHandler(Node.this);
                      private final IVerbHandler<PaxosCleanupHistory> paxosFinishPrepareCleanup = PaxosFinishPrepareCleanup.createVerbHandler(Node.this);
                      private final IVerbHandler<PaxosCleanupResponse> paxosCleanupResponse = PaxosCleanupResponse.createVerbHandler(Node.this);
                      private final IVerbHandler<PaxosCleanupComplete.Request> paxosCleanupComplete = PaxosCleanupComplete.createVerbHandler(Node.this);
                      @Override
                      public void doVerb(Message message) throws IOException
                      {
                          switch (message.verb())
                          {
                              case PAXOS2_CLEANUP_START_PREPARE_REQ:
                                  paxosStartPrepareCleanup.doVerb(message);
                                  break;
                              case PAXOS2_CLEANUP_REQ:
                                  paxosCleanupRequestIVerbHandler.doVerb(message);
                                  break;
                              case PAXOS2_CLEANUP_FINISH_PREPARE_REQ:
                                  paxosFinishPrepareCleanup.doVerb(message);
                                  break;
                              case PAXOS2_CLEANUP_RSP2:
                                  paxosCleanupResponse.doVerb(message);
                                  break;
                              case PAXOS2_CLEANUP_COMPLETE_REQ:
                                  paxosCleanupComplete.doVerb(message);
                                  break;
                              default:
                                  repairVerbHandler.doVerb(message);
                          }
                      }
                  });
  
                  activeRepairService.start();
              }
  
              public Closeable doValidation(FailingBiConsumer<ColumnFamilyStore, Validator> fn)
              {
                  FailingBiConsumer<ColumnFamilyStore, Validator> previous = this.doValidation;
                  if (previous != DEFAULT_VALIDATION)
                      throw new IllegalStateException("Attemptted to override validation, but was already overridden");
                  this.doValidation = fn;
                  return () -> this.doValidation = previous;
              }
  
              public Closeable doValidation(Function<FailingBiConsumer<ColumnFamilyStore, Validator>, FailingBiConsumer<ColumnFamilyStore, Validator>> fn)
              {
                  FailingBiConsumer<ColumnFamilyStore, Validator> previous = this.doValidation;
                  this.doValidation = fn.apply(previous);
                  return () -> this.doValidation = previous;
              }
  
              public Closeable doSync(StreamExecutor streamExecutor)
              {
                  StreamExecutor previous = this.streamExecutor;
                  if (previous != defaultStreamExecutor)
                      throw new IllegalStateException("Attemptted to override sync, but was already overridden");
                  this.streamExecutor = streamExecutor;
                  return () -> this.streamExecutor = previous;
              }
  
              void handle(Message msg)
              {
                  msg = serde(msg);
                  if (msg == null)
                  {
                      logger.warn("Got a message that failed to serialize/deserialize");
                      return;
                  }
                  for (MessageListener l : listeners)
                      l.preHandle(this, msg);
                  receiver.recieve(msg);
              }
  
              public UUID hostId()
              {
                  return hostId;
              }
  
              @Override
              public InetAddressAndPort broadcastAddressAndPort()
              {
                  return addressAndPort;
              }
  
              public Collection<Token> tokens()
              {
                  return tokens;
              }
  
              public IFailureDetector failureDetector()
              {
                  return failureDetector;
              }
  
              @Override
              public Locator locator()
              {
                  return locator;
              }
  
              @Override
              public IGossiper gossiper()
              {
                  return gossiper;
              }
  
              @Override
              public ICompactionManager compactionManager()
              {
                  return compactionManager;
              }
  
              public ExecutorFactory executorFactory()
              {
                  return globalExecutor;
              }
  
              public ScheduledExecutorPlus optionalTasks()
              {
                  return unorderedScheduled;
              }
  
              @Override
              public ScheduledExecutorPlus nonPeriodicTasks()
              {
                  return unorderedScheduled;
              }
  
              @Override
              public ScheduledExecutorPlus scheduledTasks()
              {
                  return unorderedScheduled;
              }
  
              @Override
              public Supplier<Random> random()
              {
                  return () -> rs.fork().asJdkRandom();
              }
  
              public Clock clock()
              {
                  return globalExecutor;
              }
  
              public MessageDelivery messaging()
              {
                  return messaging;
              }
  
              public MBeanWrapper mbean()
              {
                  return mbean;
              }
  
              public RepairCoordinator repair(String ks, RepairOption options)
              {
                  return repair(ks, options, true);
              }
  
              public RepairCoordinator repair(String ks, RepairOption options, boolean addFailureOnErrorNotification)
              {
                  RepairCoordinator repair = new RepairCoordinator(this, (name, tables) -> StorageService.instance.getValidColumnFamilies(false, false, name, tables), name -> StorageService.instance.getReplicas(name, broadcastAddressAndPort()), 42, options, ks);
                  if (addFailureOnErrorNotification)
                  {
                      repair.addProgressListener((tag, event) -> {
                          if (event.getType() == ProgressEventType.ERROR)
                              failures.add(new AssertionError(event.getMessage()));
                      });
                  }
                  return repair;
              }
  
              public RangesAtEndpoint getLocalReplicas(String ks)
              {
                  return StorageService.instance.getReplicas(ks, broadcastAddressAndPort());
              }
  
              public Collection<? extends Range<Token>> getPrimaryRanges(String ks)
              {
                  return StorageService.instance.getPrimaryRangesForEndpoint(ks, broadcastAddressAndPort());
              }
  
              public Collection<? extends Range<Token>> getPrimaryRangesWithinDC(String ks)
              {
                  return StorageService.instance.getPrimaryRangeForEndpointWithinDC(ks, broadcastAddressAndPort());
              }
  
              @Override
              public ActiveRepairService repair()
              {
                  return activeRepairService;
              }
  
              @Override
              public IValidationManager validationManager()
              {
                  return validationManager;
              }
  
              @Override
              public TableRepairManager repairManager(ColumnFamilyStore store)
              {
                  return new CassandraTableRepairManager(store, this)
                  {
                      @Override
                      public void snapshot(String name, Collection<Range<Token>> ranges, boolean force)
                      {
                          // no-op
                      }
                  };
              }
  
              @Override
              public StreamExecutor streamExecutor()
              {
                  return streamExecutor;
              }
  
              @Override
              public PaxosRepairState paxosRepairState()
              {
                  return paxosRepairState;
              }
  
              @Override
              public Supplier<TimeUUID> timeUUID()
              {
                  return Generators.toGen(Generators.timeUUID()).asSupplier(rs);
              }
  
              public String toString()
              {
                  return "Node{" +
                         "hostId=" + hostId +
                         ", addressAndPort=" + addressAndPort.getAddress().getHostAddress() + ":" + addressAndPort.getPort() +
                         '}';
              }
          }
  
          private Message serde(Message msg)
          {
              try (DataOutputBuffer b = DataOutputBuffer.scratchBuffer.get())
              {
                  int messagingVersion = MessagingService.current_version;
                  Message.serializer.serialize(msg, b, messagingVersion);
                  DataInputBuffer in = new DataInputBuffer(b.unsafeGetBufferAndFlip(), false);
                  return Message.serializer.deserialize(in, msg.from(), messagingVersion);
              }
              catch (Throwable e)
              {
                  failures.add(e);
                  return null;
              }
          }
      }
  
      public static class HackStrat extends LocalStrategy
      {
          public HackStrat(String keyspaceName, Map<String, String> configOptions)
          {
              super(keyspaceName, configOptions);
          }
      }
  
      /**
       * Since the clock is accessable via {@link Global#currentTimeMillis()} and {@link Global#nanoTime()}, and only repair subsystem has the requirement not to touch that, this class
       * acts as a safty check that validates that repair does not touch these methods but allows the other subsystems to do so.
       */
      public static class ClockAccess implements Clock
      {
          private static final Set<Thread> OWNERS = Collections.synchronizedSet(new HashSet<>());
          private final Clock delegate = new Default();
  
          public static void includeThreadAsOwner()
          {
              OWNERS.add(Thread.currentThread());
          }
  
          @Override
          public long nanoTime()
          {
              checkAccess();
              return delegate.nanoTime();
          }
  
          @Override
          public long currentTimeMillis()
          {
              checkAccess();
              return delegate.currentTimeMillis();
          }
  
          private enum Access
          {MAIN_THREAD_ONLY, REJECT, IGNORE}
  
          private void checkAccess()
          {
              Access access = StackWalker.getInstance().walk(frames -> {
                  Iterator<StackWalker.StackFrame> it = frames.iterator();
                  boolean topLevel = false;
                  while (it.hasNext())
                  {
                      StackWalker.StackFrame next = it.next();
                      if (!topLevel)
                      {
                          // need to find the top level!
                          while (!Clock.Global.class.getName().equals(next.getClassName()))
                          {
                              assert it.hasNext();
                              next = it.next();
                          }
                          topLevel = true;
                          assert it.hasNext();
                          next = it.next();
                      }
                      if (FuzzTestBase.class.getName().equals(next.getClassName())) return Access.MAIN_THREAD_ONLY;
  
                      // this is non-deterministic... but since the scope of the work is testing repair and not paxos... this is unblocked for now...
                      if (("org.apache.cassandra.service.paxos.Paxos".equals(next.getClassName()) && "newBallot".equals(next.getMethodName()))
                          || ("org.apache.cassandra.service.paxos.uncommitted.PaxosBallotTracker".equals(next.getClassName()) && "updateLowBound".equals(next.getMethodName())))
                          return Access.MAIN_THREAD_ONLY;
                      if (next.getClassName().startsWith("org.apache.cassandra.accord.")
                          || next.getClassName().startsWith("org.apache.cassandra.journal.")
                          || next.getClassName().startsWith("org.apache.cassandra.service.accord.")
                          || next.getClassName().startsWith("org.apache.cassandra.db.")
                          || next.getClassName().startsWith("org.apache.cassandra.gms.")
                          || next.getClassName().startsWith("org.apache.cassandra.cql3.")
                          || next.getClassName().startsWith("org.apache.cassandra.metrics.")
                          || next.getClassName().startsWith("org.apache.cassandra.utils.concurrent.")
                          || next.getClassName().startsWith("org.apache.cassandra.tcm")
                          || next.getClassName().startsWith("org.apache.cassandra.utils.TimeUUID") // this would be good to solve
                          || next.getClassName().startsWith("org.apache.cassandra.schema")
                          || next.getClassName().startsWith(PendingAntiCompaction.class.getName()))
                          return Access.IGNORE;
                      if (next.getClassName().startsWith("org.apache.cassandra.repair") || ActiveRepairService.class.getName().startsWith(next.getClassName()))
                          return Access.REJECT;
                  }
                  return Access.IGNORE;
              });
              Thread current = Thread.currentThread();
              switch (access)
              {
                  case IGNORE:
                      return;
                  case REJECT:
                      throw new IllegalStateException("Rejecting access");
                  case MAIN_THREAD_ONLY:
                      if (!OWNERS.contains(current)) throw new IllegalStateException("Accessed in wrong thread: " + current);
                      break;
              }
          }
      }
  
      static class SimulatedFault extends RuntimeException
      {
          SimulatedFault(String message)
          {
              super(message);
          }
      }
  }
  