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.distributed.test;
  
  import java.net.InetSocketAddress;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicIntegerArray;
  import java.util.function.Consumer;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Iterators;
  import com.google.common.collect.Sets;
  import com.google.common.util.concurrent.Uninterruptibles;
  import org.junit.Assert;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.CassandraRelevantProperties;
  import org.apache.cassandra.config.Config;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.cql3.QueryOptions;
  import org.apache.cassandra.cql3.QueryProcessor;
  import org.apache.cassandra.cql3.statements.SelectStatement;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.ReadExecutionController;
  import org.apache.cassandra.db.ReadQuery;
  import org.apache.cassandra.db.SystemKeyspace;
  import org.apache.cassandra.db.compaction.CompactionManager;
  import org.apache.cassandra.db.marshal.Int32Type;
  import org.apache.cassandra.db.partitions.PartitionIterator;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.db.rows.RowIterator;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.Constants;
  import org.apache.cassandra.distributed.api.ConsistencyLevel;
  import org.apache.cassandra.distributed.api.Feature;
  import org.apache.cassandra.distributed.api.ICluster;
  import org.apache.cassandra.distributed.api.IInstance;
  import org.apache.cassandra.distributed.api.IInstanceConfig;
  import org.apache.cassandra.distributed.api.IInvokableInstance;
  import org.apache.cassandra.distributed.api.IMessageFilters;
  import org.apache.cassandra.gms.FailureDetector;
  import org.apache.cassandra.distributed.api.TokenSupplier;
  import org.apache.cassandra.distributed.shared.NetworkTopology;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.repair.RepairParallelism;
  import org.apache.cassandra.repair.SharedContext;
  import org.apache.cassandra.repair.messages.RepairOption;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.service.ActiveRepairService;
  import org.apache.cassandra.service.ClientState;
  import org.apache.cassandra.service.StorageService;
  import org.apache.cassandra.service.paxos.Ballot;
  import org.apache.cassandra.service.paxos.Commit;
  import org.apache.cassandra.service.paxos.PaxosState;
  import org.apache.cassandra.service.paxos.cleanup.PaxosCleanup;
  import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
  import org.apache.cassandra.streaming.PreviewKind;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.tcm.ClusterMetadataService;
  import org.apache.cassandra.tcm.membership.Directory;
  import org.apache.cassandra.tcm.membership.NodeVersion;
  import org.apache.cassandra.tcm.transformations.ForceSnapshot;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.CassandraVersion;
  import org.apache.cassandra.utils.ExecutorUtils;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.Pair;
  import org.apache.cassandra.utils.TimeUUID;
  
  import static java.util.concurrent.TimeUnit.SECONDS;
  import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
  import static org.apache.cassandra.distributed.shared.AssertUtils.row;
  import static org.apache.cassandra.net.Verb.PAXOS2_CLEANUP_FINISH_PREPARE_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS2_CLEANUP_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS2_COMMIT_AND_PREPARE_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS2_PREPARE_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS2_PREPARE_RSP;
  import static org.apache.cassandra.net.Verb.PAXOS2_PROPOSE_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS2_PROPOSE_RSP;
  import static org.apache.cassandra.net.Verb.PAXOS2_REPAIR_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_REQ;
  import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_RSP;
  import static org.apache.cassandra.schema.SchemaConstants.SYSTEM_KEYSPACE_NAME;
  
  public class PaxosRepairTest extends TestBaseImpl
  {
      private static final Logger logger = LoggerFactory.getLogger(PaxosRepairTest.class);
      private static final String TABLE = "tbl";
  
      static
      {
          CassandraRelevantProperties.PAXOS_USE_SELF_EXECUTION.setBoolean(false);
          CassandraRelevantProperties.TCM_USE_ATOMIC_LONG_PROCESSOR.setBoolean(true);
          CassandraRelevantProperties.TCM_ALLOW_TRANSFORMATIONS_DURING_UPGRADES.setBoolean(true); // for paxosRepairVersionGate
          DatabaseDescriptor.daemonInitialization();
      }
  
      private static int getUncommitted(IInvokableInstance instance, String keyspace, String table)
      {
          if (instance.isShutdown())
              return 0;
          int uncommitted = instance.callsOnInstance(() -> {
              TableMetadata meta = Schema.instance.getTableMetadata(keyspace, table);
              return Iterators.size(PaxosState.uncommittedTracker().uncommittedKeyIterator(meta.id, null));
          }).call();
          logger.info("{} has {} uncommitted instances", instance, uncommitted);
          return uncommitted;
      }
  
      private static void assertAllAlive(Cluster cluster)
      {
          Set<InetAddressAndPort> allEndpoints = cluster.stream().map(i -> InetAddressAndPort.getByAddress(i.broadcastAddress())).collect(Collectors.toSet());
          cluster.stream().forEach(instance -> {
              instance.runOnInstance(() -> {
                  ImmutableSet<InetAddressAndPort> endpoints = ImmutableSet.copyOf(ClusterMetadata.current().directory.allJoinedEndpoints());
                  Assert.assertEquals(allEndpoints, endpoints);
                  for (InetAddressAndPort endpoint : endpoints)
                      Assert.assertTrue(FailureDetector.instance.isAlive(endpoint));
              });
          });
      }
  
      private static void assertUncommitted(IInvokableInstance instance, String ks, String table, int expected)
      {
          Assert.assertEquals(expected, getUncommitted(instance, ks, table));
      }
  
      private static boolean hasUncommitted(Cluster cluster, String ks, String table)
      {
          return cluster.stream().map(instance -> getUncommitted(instance, ks, table)).reduce((a, b) -> a + b).get() > 0;
      }
  
      private static boolean hasUncommittedQuorum(Cluster cluster, String ks, String table)
      {
          int uncommitted = 0;
          for (int i=0; i<cluster.size(); i++)
          {
              if (getUncommitted(cluster.get(i+1), ks, table) > 0)
                  uncommitted++;
          }
          return uncommitted >= ((cluster.size() / 2) + 1);
      }
  
      private static void repair(Cluster cluster, String keyspace, String table, boolean force)
      {
          Map<String, String> options = new HashMap<>();
          options.put(RepairOption.PARALLELISM_KEY, RepairParallelism.SEQUENTIAL.getName());
          options.put(RepairOption.PRIMARY_RANGE_KEY, Boolean.toString(false));
          options.put(RepairOption.INCREMENTAL_KEY, Boolean.toString(false));
          options.put(RepairOption.JOB_THREADS_KEY, Integer.toString(1));
          options.put(RepairOption.TRACE_KEY, Boolean.toString(false));
          options.put(RepairOption.COLUMNFAMILIES_KEY, "");
          options.put(RepairOption.PULL_REPAIR_KEY, Boolean.toString(false));
          options.put(RepairOption.FORCE_REPAIR_KEY, Boolean.toString(force));
          options.put(RepairOption.PREVIEW, PreviewKind.NONE.toString());
          options.put(RepairOption.IGNORE_UNREPLICATED_KS, Boolean.toString(false));
          options.put(RepairOption.REPAIR_DATA_KEY, Boolean.toString(false));
          options.put(RepairOption.REPAIR_PAXOS_KEY, Boolean.toString(true));
          options.put(RepairOption.REPAIR_ACCORD_KEY, Boolean.toString(false));
  
          cluster.get(1).runOnInstance(() -> {
              int cmd = StorageService.instance.repairAsync(keyspace, options);
  
              while (true)
              {
                  try
                  {
                      Thread.sleep(500);
                  }
                  catch (InterruptedException e)
                  {
                      throw new AssertionError(e);
                  }
                  Pair<ActiveRepairService.ParentRepairStatus, List<String>> status = ActiveRepairService.instance().getRepairStatus(cmd);
                  if (status == null)
                      continue;
  
                  switch (status.left)
                  {
                      case IN_PROGRESS:
                          continue;
                      case COMPLETED:
                          return;
                      default:
                          throw new AssertionError("Repair failed with errors: " + status.right);
                  }
              }
          });
      }
  
      private static void repair(Cluster cluster, String keyspace, String table)
      {
          repair(cluster, keyspace, table, false);
      }
  
      private static final Consumer<IInstanceConfig> WITH_NETWORK = cfg -> {
          cfg.with(Feature.NETWORK);
          cfg.with(Feature.GOSSIP);
          cfg.set("paxos_purge_grace_period", "0s");
          cfg.set("paxos_state_purging", Config.PaxosStatePurging.repaired.toString());
          cfg.set("paxos_variant", "v2_without_linearizable_reads");
          cfg.set("truncate_request_timeout", "1000ms");
          cfg.set("partitioner", "ByteOrderedPartitioner");
          cfg.set("initial_token", ByteBufferUtil.bytesToHex(ByteBufferUtil.bytes(cfg.num() * 100)));
      };
  
      private static final Consumer<IInstanceConfig> WITHOUT_NETWORK = cfg -> {
          cfg.set("paxos_purge_grace_period", "0s");
          cfg.set("paxos_state_purging", Config.PaxosStatePurging.repaired.toString());
          cfg.set("paxos_variant", "v2_without_linearizable_reads");
          cfg.set("truncate_request_timeout", "1000ms");
          cfg.set("partitioner", "ByteOrderedPartitioner");
          cfg.set("initial_token", ByteBufferUtil.bytesToHex(ByteBufferUtil.bytes(cfg.num() * 100)));
      };
  
      @Test
      public void paxosRepairTest() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = init(Cluster.build(3).withConfig(WITH_NETWORK).withoutVNodes().start()))
          {
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
              cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
              Assert.assertFalse(hasUncommittedQuorum(cluster, KEYSPACE, TABLE));
  
              assertAllAlive(cluster);
              cluster.verbs(PAXOS_COMMIT_REQ).drop();
              try
              {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }
              catch (RuntimeException e)
              {
                  // exception expected
              }
  
              Assert.assertTrue(hasUncommitted(cluster, KEYSPACE, TABLE));
  
              cluster.filters().reset();
  
              assertAllAlive(cluster);
              repair(cluster, KEYSPACE, TABLE);
  
              Assert.assertFalse(hasUncommitted(cluster, KEYSPACE, TABLE));
  
              cluster.forEach(i -> i.runOnInstance(() -> {
                  compactPaxos();
                  Map<Integer, PaxosRow> rows = getPaxosRows();
                  assertLowBoundPurged(rows.values());
                  Assert.assertEquals(Sets.newHashSet(400), rows.keySet());
              }));
  
              // check that operations occuring after the last repair are not purged
              cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (500, 3, 3) IF NOT EXISTS", ConsistencyLevel.QUORUM);
              cluster.forEach(i -> i.runOnInstance(() -> {
                  compactPaxos();
                  Map<Integer, PaxosRow> rows = getPaxosRows();
                  assertLowBoundPurged(rows.values());
                  Assert.assertEquals(Sets.newHashSet(400, 500), rows.keySet());
              }));
          }
      }
  
      @Test
      public void topologyChangePaxosTest() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = builder().withNodes(3)
                                          .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4))
                                          .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
                                          .withConfig(WITH_NETWORK)
                                          .withoutVNodes()
                                          .start())
          {
              init(cluster);
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
              cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
  
              cluster.verbs(PAXOS_COMMIT_REQ).drop();
              try
              {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (350, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }
              catch (RuntimeException e)
              {
                  // exception expected
              }
              Assert.assertTrue(hasUncommitted(cluster, KEYSPACE, TABLE));
  
              cluster.filters().reset();
  
              // node 4 starting should repair paxos and inform the other nodes of its gossip state
              IInstanceConfig config = cluster.newInstanceConfig()
                                              .set("auto_bootstrap", true)
                                              .set(Constants.KEY_DTEST_FULL_STARTUP, true);
              IInvokableInstance node4 = cluster.bootstrap(config);
              node4.startup();
              Assert.assertFalse(hasUncommittedQuorum(cluster, KEYSPACE, TABLE));
          }
      }
  
      @Test
      public void paxosCleanupWithReproposal() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = init(Cluster.build(3)
                                             .withConfig(cfg -> cfg
                                                                .set("paxos_variant", "v2")
                                                                .set("paxos_purge_grace_period", "0s")
                                                                .set("paxos_state_purging", Config.PaxosStatePurging.repaired.toString())
                                                                .set("truncate_request_timeout", "1000ms"))
                                             .withoutVNodes()
                                             .start()))
          {
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
  
              cluster.verbs(PAXOS_COMMIT_REQ).drop();
              try
              {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }
              catch (RuntimeException e)
              {
                  // exception expected
              }
              Assert.assertTrue(hasUncommitted(cluster, KEYSPACE, TABLE));
              cluster.forEach(i -> i.runOnInstance(() -> Keyspace.open("system").getColumnFamilyStore("paxos").forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS)));
  
              CountDownLatch haveFetchedLowBound = new CountDownLatch(1);
              CountDownLatch haveReproposed = new CountDownLatch(1);
              cluster.verbs(PAXOS2_CLEANUP_FINISH_PREPARE_REQ).inbound().messagesMatching((from, to, verb) -> {
                  haveFetchedLowBound.countDown();
                  Uninterruptibles.awaitUninterruptibly(haveReproposed);
                  return false;
              }).drop();
  
              ExecutorService executor = Executors.newCachedThreadPool();
              List<InetAddressAndPort> endpoints = cluster.stream().map(IInstance::broadcastAddress).map(InetAddressAndPort::getByAddress).collect(Collectors.toList());
              Future<?> cleanup = cluster.get(1).appliesOnInstance((List<? extends InetSocketAddress> es, ExecutorService exec)-> {
                  TableMetadata metadata = Keyspace.open(KEYSPACE).getMetadata().getTableOrViewNullable(TABLE);
                  return PaxosCleanup.cleanup(SharedContext.Global.instance, es.stream().map(InetAddressAndPort::getByAddress).collect(Collectors.toSet()), metadata, StorageService.instance.getLocalRanges(KEYSPACE), false, exec);
              }).apply(endpoints, executor);
  
              Uninterruptibles.awaitUninterruptibly(haveFetchedLowBound);
              IMessageFilters.Filter filter2 = cluster.verbs(PAXOS_COMMIT_REQ, PAXOS2_COMMIT_AND_PREPARE_REQ).drop();
              try
              {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }
              catch (RuntimeException e)
              {
                  // exception expected
              }
              filter2.off();
              haveReproposed.countDown();
              cluster.filters().reset();
  
              cleanup.get();
              ExecutorUtils.shutdownNowAndWait(1L, TimeUnit.MINUTES, executor);
              Assert.assertFalse(hasUncommitted(cluster, KEYSPACE, TABLE));
              cluster.forEach(i -> i.runOnInstance(PaxosRepairTest::compactPaxos));
              for (int i = 1 ; i <= 3 ; ++i)
                  assertRows(cluster.get(i).executeInternal("SELECT * FROM " + KEYSPACE + '.' + TABLE + " WHERE pk = 1"), row(1, 1, 1));
  
              Assert.assertFalse(hasUncommittedQuorum(cluster, KEYSPACE, TABLE));
              assertLowBoundPurged(cluster);
          }
      }
  
      @SuppressWarnings("unused")
      @Test
      public void paxosCleanupWithReproposalClashingTimestamp() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = init(Cluster.build(5)
                                             .withConfig(cfg -> cfg
                                                                .set("paxos_variant", "v2")
                                                                .set("paxos_purge_grace_period", "0s")
                                                                .set("paxos_cache_size", "0MiB")
                                                                .set("truncate_request_timeout", "1000ms"))
                                             .withoutVNodes()
                                             .start()))
          {
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
  
              // we ensure:
              //  - node 1 only witnesses a promise that conflicts with something we committed
              //  - node 2 does not witness the commit, so it has an in progress proposal
              //  - node 3 does not witness the proposal, so that we have an incomplete commit
              //  - node 1's response arrives first, so that it might retain its promise as latestWitnessed (without bugfix)
  
              CountDownLatch haveStartedCleanup = new CountDownLatch(1);
              CountDownLatch haveInsertedClashingPromise = new CountDownLatch(1);
              IMessageFilters.Filter pauseCleanupUntilCommitted = cluster.verbs(PAXOS2_CLEANUP_REQ).from(1).to(1).outbound().messagesMatching((from, to, verb) -> {
                  haveStartedCleanup.countDown();
                  Uninterruptibles.awaitUninterruptibly(haveInsertedClashingPromise);
                  return false;
              }).drop();
  
              ExecutorService executor = Executors.newCachedThreadPool();
              List<InetAddressAndPort> endpoints = cluster.stream().map(i -> InetAddressAndPort.getByAddress(i.broadcastAddress())).collect(Collectors.toList());
              Future<?> cleanup = cluster.get(1).appliesOnInstance((List<? extends InetSocketAddress> es, ExecutorService exec)-> {
                  TableMetadata metadata = Keyspace.open(KEYSPACE).getMetadata().getTableOrViewNullable(TABLE);
                  return PaxosCleanup.cleanup(SharedContext.Global.instance, es.stream().map(InetAddressAndPort::getByAddress).collect(Collectors.toSet()), metadata, StorageService.instance.getLocalRanges(KEYSPACE), false, exec);
              }).apply(endpoints, executor);
  
              IMessageFilters.Filter dropAllTo1 = cluster.verbs(PAXOS2_PREPARE_REQ, PAXOS2_PROPOSE_REQ, PAXOS_COMMIT_REQ).from(2).to(1).outbound().drop();
              IMessageFilters.Filter dropCommitTo3 = cluster.verbs(PAXOS_COMMIT_REQ).from(2).to(3).outbound().drop();
              IMessageFilters.Filter dropAcceptTo4 = cluster.verbs(PAXOS2_PROPOSE_REQ).from(2).to(4).outbound().drop();
  
              CountDownLatch haveFetchedClashingRepair = new CountDownLatch(1);
              AtomicIntegerArray fetchResponseIds = new AtomicIntegerArray(new int[] { -1, -1, -1, -1, -1, -1 });
              cluster.verbs(PAXOS2_REPAIR_REQ).outbound().from(1).messagesMatching((from, to, msg) -> {
                  fetchResponseIds.set(to, msg.id());
                  return false;
              }).drop();
              cluster.verbs(PAXOS2_PREPARE_RSP, PAXOS2_PROPOSE_RSP, PAXOS_COMMIT_RSP).outbound().to(1).messagesMatching((from, to, msg) -> {
                  if (fetchResponseIds.get(from) == msg.id())
                  {
                      if (from == 1) haveFetchedClashingRepair.countDown();
                      else Uninterruptibles.awaitUninterruptibly(haveFetchedClashingRepair);
                  }
                  return false;
              }).drop();
  
              Uninterruptibles.awaitUninterruptibly(haveStartedCleanup);
              cluster.coordinator(2).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.ONE);
  
              UUID cfId = cluster.get(2).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE).metadata.id.asUUID());
              TimeUUID uuid = (TimeUUID) cluster.get(2).executeInternal("select in_progress_ballot from system.paxos WHERE row_key = ? and cf_id = ?", Int32Type.instance.decompose(1), cfId)[0][0];
              TimeUUID clashingUuid = TimeUUID.fromBytes(uuid.msb(), 0);
              cluster.get(1).executeInternal("update system.paxos set in_progress_ballot = ? WHERE row_key = ? and cf_id = ?", clashingUuid, Int32Type.instance.decompose(1), cfId);
              Assert.assertEquals(clashingUuid, cluster.get(1).executeInternal("select in_progress_ballot from system.paxos WHERE row_key = ? and cf_id = ?", Int32Type.instance.decompose(1), cfId)[0][0]);
  
              Assert.assertTrue(hasUncommitted(cluster, KEYSPACE, TABLE));
              haveInsertedClashingPromise.countDown();
  
              cleanup.get();
              ExecutorUtils.shutdownNowAndWait(1L, TimeUnit.MINUTES, executor);
          }
      }
  
      @Test
      public void paxosCleanupWithDelayedProposal() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = init(Cluster.build(3)
                                             .withConfig(cfg -> cfg
                                                                .set("paxos_variant", "v2")
                                                                .set("paxos_purge_grace_period", "0s")
                                                                .set("paxos_state_purging", Config.PaxosStatePurging.repaired.toString())
                                                                .set("truncate_request_timeout", "1000ms"))
                                             .withoutVNodes()
                                             .start())
          )
          {
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
  
              CountDownLatch haveFinishedRepair = new CountDownLatch(1);
              cluster.verbs(PAXOS2_PREPARE_REQ).messagesMatching((from, to, verb) -> {
                  Uninterruptibles.awaitUninterruptibly(haveFinishedRepair);
                  return false;
              }).drop();
              cluster.verbs(PAXOS_COMMIT_REQ).drop();
              Future<?> insert = cluster.get(1).async(() -> {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }).call();
              cluster.verbs(PAXOS2_CLEANUP_FINISH_PREPARE_REQ).messagesMatching((from, to, verb) -> {
                  haveFinishedRepair.countDown();
                  try { insert.get(); } catch (Throwable t) {}
                  cluster.filters().reset();
                  return false;
              }).drop();
  
              ExecutorService executor = Executors.newCachedThreadPool();
  
              Uninterruptibles.sleepUninterruptibly(10L, TimeUnit.MILLISECONDS);
  
              List<InetAddressAndPort> endpoints = cluster.stream().map(i -> InetAddressAndPort.getByAddress(i.broadcastAddress())).collect(Collectors.toList());
              Future<?> cleanup = cluster.get(1).appliesOnInstance((List<? extends InetSocketAddress> es, ExecutorService exec)-> {
                  TableMetadata metadata = Keyspace.open(KEYSPACE).getMetadata().getTableOrViewNullable(TABLE);
                  return PaxosCleanup.cleanup(SharedContext.Global.instance, es.stream().map(InetAddressAndPort::getByAddress).collect(Collectors.toSet()), metadata, StorageService.instance.getLocalRanges(KEYSPACE), false, exec);
              }).apply(endpoints, executor);
  
              cleanup.get();
              try
              {
                  insert.get();
              }
              catch (Throwable t)
              {
              }
              ExecutorUtils.shutdownNowAndWait(1L, TimeUnit.MINUTES, executor);
              Assert.assertFalse(hasUncommittedQuorum(cluster, KEYSPACE, TABLE));
  
              assertLowBoundPurged(cluster);
          }
      }
  
      private static void setVersion(ICluster<IInvokableInstance> cluster, InetSocketAddress peer, String version)
      {
          cluster.get(1).acceptsOnInstance((InetSocketAddress addr) -> {
              ClusterMetadata cm = ClusterMetadata.current();
              Directory directory = cm.directory.withNodeVersion(cm.directory.peerId(InetAddressAndPort.getByAddress(addr)),
                                                                 new NodeVersion(new CassandraVersion(version), NodeVersion.CURRENT_METADATA_VERSION));
  
              ClusterMetadata nextMetadata = cm.transformer().with(directory).build().metadata;
              ClusterMetadataService.instance().commit(new ForceSnapshot(nextMetadata));
          }).accept(cluster.get(2).broadcastAddress());
      }
  
      private static void assertRepairFailsWithVersion(Cluster cluster, String version)
      {
          setVersion(cluster, cluster.get(2).broadcastAddress(), version);
  
          try
          {
              repair(cluster, KEYSPACE, TABLE);
          }
          catch (AssertionError e)
          {
              return;
          }
          Assert.fail("Repair should have failed on unsupported version");
      }
  
      private static void assertRepairSucceedsWithVersion(Cluster cluster, String version)
      {
          setVersion(cluster, cluster.get(2).broadcastAddress(), version);
          repair(cluster, KEYSPACE, TABLE);
      }
  
      @Test
      public void paxosRepairVersionGate() throws Throwable
      {
          // TODO: fails with vnode enabled
          try (Cluster cluster = init(Cluster.build(3).withConfig(WITHOUT_NETWORK).withoutVNodes().start()))
          {
              cluster.schemaChange("CREATE TABLE " + KEYSPACE + '.' + TABLE + " (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
              cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
              Assert.assertFalse(hasUncommittedQuorum(cluster, KEYSPACE, TABLE));
  
              assertAllAlive(cluster);
              cluster.verbs(PAXOS_COMMIT_REQ).drop();
              try
              {
                  cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + '.' + TABLE + " (pk, ck, v) VALUES (400, 2, 2) IF NOT EXISTS", ConsistencyLevel.QUORUM);
                  Assert.fail("expected write timeout");
              }
              catch (RuntimeException e)
              {
                  // exception expected
              }
  
              Assert.assertTrue(hasUncommitted(cluster, KEYSPACE, TABLE));
  
              cluster.filters().reset();
  
              assertAllAlive(cluster);
  
              assertRepairFailsWithVersion(cluster, "3.0.24");
              assertRepairFailsWithVersion(cluster, "4.0.0");
  
              // test valid versions
              assertRepairSucceedsWithVersion(cluster, "4.1.0");
          }
      }
  
      private static class PaxosRow
      {
          final DecoratedKey key;
          final Row row;
  
          PaxosRow(DecoratedKey key, Row row)
          {
              this.key = key;
              this.row = row;
          }
  
          public String toString()
          {
              TableMetadata table = Schema.instance.getTableMetadata(SYSTEM_KEYSPACE_NAME, SystemKeyspace.PAXOS);
              return ByteBufferUtil.bytesToHex(key.getKey()) + " -> " + row.toString(table, true);
          }
      }
  
      private static void compactPaxos()
      {
          ColumnFamilyStore paxos = Keyspace.open(SYSTEM_KEYSPACE_NAME).getColumnFamilyStore(SystemKeyspace.PAXOS);
          FBUtilities.waitOnFuture(paxos.forceFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS));
          FBUtilities.waitOnFutures(CompactionManager.instance.submitMaximal(paxos, 0, false, 0));
      }
  
      private static Map<Integer, PaxosRow> getPaxosRows()
      {
          Map<Integer, PaxosRow> rows = new HashMap<>();
          String queryStr = "SELECT * FROM " + SYSTEM_KEYSPACE_NAME + '.' + SystemKeyspace.PAXOS;
          SelectStatement stmt = (SelectStatement) QueryProcessor.parseStatement(queryStr).prepare(ClientState.forInternalCalls());
          ReadQuery query = stmt.getQuery(QueryOptions.DEFAULT, FBUtilities.nowInSeconds());
          try (ReadExecutionController controller = query.executionController(); PartitionIterator partitions = query.executeInternal(controller))
          {
              while (partitions.hasNext())
              {
                  try (RowIterator partition = partitions.next())
                  {
                      while (partition.hasNext())
                      {
                          rows.put(Int32Type.instance.compose(partition.partitionKey().getKey()),
                                   new PaxosRow(partition.partitionKey(), partition.next()));
                      }
                  }
              }
          }
          return rows;
      }
  
      private static void assertLowBoundPurged(Collection<PaxosRow> rows)
      {
          Assert.assertEquals(0, DatabaseDescriptor.getPaxosPurgeGrace(SECONDS));
          String ip = FBUtilities.getBroadcastAddressAndPort().toString();
          for (PaxosRow row : rows)
          {
              Ballot keyLowBound = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE).getPaxosRepairLowBound(row.key);
              Assert.assertTrue(ip, Commit.isAfter(keyLowBound, Ballot.none()));
              Assert.assertFalse(ip, PaxosRows.hasBallotBeforeOrEqualTo(row.row, keyLowBound));
          }
      }
  
      private static void assertLowBoundPurged(IInvokableInstance instance)
      {
          instance.runOnInstance(() -> assertLowBoundPurged(getPaxosRows().values()));
      }
  
      private static void assertLowBoundPurged(Cluster cluster)
      {
          cluster.forEach(PaxosRepairTest::assertLowBoundPurged);
      }
  }
  