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.service.paxos.cleanup;
  
  import java.util.Collection;
  import java.util.Map;
  import java.util.UUID;
  import java.util.concurrent.ConcurrentHashMap;
  
  import com.google.common.base.Preconditions;
  import com.google.common.util.concurrent.Uninterruptibles;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.ConsistencyLevel;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.repair.SharedContext;
  import org.apache.cassandra.schema.Schema;
  import org.apache.cassandra.schema.TableId;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.service.paxos.AbstractPaxosRepair;
  import org.apache.cassandra.service.paxos.PaxosRepair;
  import org.apache.cassandra.service.paxos.PaxosState;
  import org.apache.cassandra.service.paxos.uncommitted.UncommittedPaxosKey;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.utils.Clock;
  import org.apache.cassandra.utils.CloseableIterator;
  import org.apache.cassandra.utils.concurrent.AsyncFuture;
  
  import static java.util.concurrent.TimeUnit.MICROSECONDS;
  import static java.util.concurrent.TimeUnit.MILLISECONDS;
  import static java.util.concurrent.TimeUnit.SECONDS;
  import static org.apache.cassandra.config.DatabaseDescriptor.getCasContentionTimeout;
  import static org.apache.cassandra.config.DatabaseDescriptor.getWriteRpcTimeout;
  import static org.apache.cassandra.service.paxos.cleanup.PaxosCleanupSession.TIMEOUT_NANOS;
  
  public class PaxosCleanupLocalCoordinator extends AsyncFuture<PaxosCleanupResponse>
  {
      private static final Logger logger = LoggerFactory.getLogger(PaxosCleanupLocalCoordinator.class);
      private static final UUID INTERNAL_SESSION = new UUID(0, 0);
  
      private final UUID session;
      private final TableId tableId;
      private final TableMetadata table;
      private final Collection<Range<Token>> ranges;
      private final CloseableIterator<UncommittedPaxosKey> uncommittedIter;
      private final SharedContext ctx;
      private int count = 0;
      private final long deadline;
      private final boolean autoRepair;
  
      private final Map<DecoratedKey, AbstractPaxosRepair> inflight = new ConcurrentHashMap<>();
      private final PaxosTableRepairs tableRepairs;
  
      private PaxosCleanupLocalCoordinator(SharedContext ctx, UUID session, TableId tableId, Collection<Range<Token>> ranges, CloseableIterator<UncommittedPaxosKey> uncommittedIter, boolean autoRepair)
      {
          this.ctx = ctx;
          this.session = session;
          this.tableId = tableId;
          this.table = Schema.instance.getTableMetadata(tableId);
          this.ranges = ranges;
          this.uncommittedIter = uncommittedIter;
          this.tableRepairs = ctx.paxosRepairState().getForTable(tableId);
          this.deadline = TIMEOUT_NANOS + ctx.clock().nanoTime();
          this.autoRepair = autoRepair;
      }
  
      public synchronized void start()
      {
          if (table == null)
          {
              fail("Unknown tableId: " + tableId);
              return;
          }
  
          ClusterMetadata metadata = ClusterMetadata.current();
          if (metadata.schema.getKeyspace(table.keyspace) == null)
          {
              fail("Unknown keyspace: " + table.keyspace);
              return;
          }
  
          if (!PaxosRepair.validatePeerCompatibility(ctx, metadata, table, ranges))
          {
              fail("Unsupported peer versions for " + tableId + ' ' + ranges.toString());
              return;
          }
  
          if (autoRepair)
              logger.debug("Completing uncommitted paxos instances for {} on ranges {} for session {}", table, ranges, session);
          else
              logger.info("Completing uncommitted paxos instances for {} on ranges {} for session {}", table, ranges, session);
  
          scheduleKeyRepairsOrFinish();
      }
  
      public static PaxosCleanupLocalCoordinator create(SharedContext ctx, PaxosCleanupRequest request)
      {
          CloseableIterator<UncommittedPaxosKey> iterator = PaxosState.uncommittedTracker().uncommittedKeyIterator(request.tableId, request.ranges);
          return new PaxosCleanupLocalCoordinator(ctx, request.session, request.tableId, request.ranges, iterator, false);
      }
  
      public static PaxosCleanupLocalCoordinator createForAutoRepair(SharedContext ctx, TableId tableId, Collection<Range<Token>> ranges)
      {
          CloseableIterator<UncommittedPaxosKey> iterator = PaxosState.uncommittedTracker().uncommittedKeyIterator(tableId, ranges);
          return new PaxosCleanupLocalCoordinator(ctx, INTERNAL_SESSION, tableId, ranges, iterator, true);
      }
  
      /**
       * Schedule as many key repairs as we can, up to the paralellism limit. If no repairs are scheduled and
       * none are in flight when the iterator is exhausted, the session will be finished
       */
      private void scheduleKeyRepairsOrFinish()
      {
          int parallelism = DatabaseDescriptor.getPaxosRepairParallelism();
          Preconditions.checkArgument(parallelism > 0);
          if (inflight.size() < parallelism)
          {
              if (ctx.clock().nanoTime() - deadline >= 0)
              {
                  fail("timeout");
                  return;
              }
  
              long txnTimeoutMicros = Math.max(getCasContentionTimeout(MICROSECONDS), getWriteRpcTimeout(MICROSECONDS));
              boolean waitForCoordinator = DatabaseDescriptor.getPaxosRepairRaceWait();
              while (inflight.size() < parallelism && uncommittedIter.hasNext())
                  repairKey(uncommittedIter.next(), txnTimeoutMicros, waitForCoordinator);
  
          }
  
          if (inflight.isEmpty())
              finish();
      }
  
      private boolean repairKey(UncommittedPaxosKey uncommitted, long txnTimeoutMicros, boolean waitForCoordinator)
      {
          logger.trace("repairing {}", uncommitted);
          Preconditions.checkState(!inflight.containsKey(uncommitted.getKey()));
          ConsistencyLevel consistency = uncommitted.getConsistencyLevel();
  
          // we don't know the consistency of this operation, presumably because it originated
          // before we started tracking paxos cl, so we don't attempt to repair it
          if (consistency == null)
              return false;
  
          if (waitForCoordinator)
              maybeWaitForOriginalCoordinator(uncommitted, txnTimeoutMicros);
  
          inflight.put(uncommitted.getKey(), tableRepairs.startOrGetOrQueue(uncommitted.getKey(), uncommitted.ballot(), uncommitted.getConsistencyLevel(), table, result -> {
              if (result.wasSuccessful())
                  onKeyFinish(uncommitted.getKey());
              else
                  onKeyFailure(result.toString());
          }));
          return true;
      }
  
      /**
       * Wait to repair things that are still potentially executing at the original coordinator to avoid
       * causing timeouts. This should only have to happen at most a few times when the repair starts
       */
      private static void maybeWaitForOriginalCoordinator(UncommittedPaxosKey uncommitted, long txnTimeoutMicros)
      {
          long nowMicros = MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
          long ballotElapsedMicros = nowMicros - uncommitted.ballot().unixMicros();
          if (ballotElapsedMicros < 0 && Math.abs(ballotElapsedMicros) > SECONDS.toMicros(1))
              logger.warn("Encountered ballot that is more than 1 second in the future, is there a clock sync issue? {}", uncommitted.ballot());
          if (ballotElapsedMicros < txnTimeoutMicros)
          {
              long sleepMicros = txnTimeoutMicros - ballotElapsedMicros;
              logger.info("Paxos auto repair encountered a potentially in progress ballot, sleeping {}us to allow the in flight operation to finish", sleepMicros);
              Uninterruptibles.sleepUninterruptibly(sleepMicros, MICROSECONDS);
          }
      }
  
      private synchronized void onKeyFinish(DecoratedKey key)
      {
          if (!inflight.containsKey(key))
              return;
          logger.trace("finished repairing {}", key);
          inflight.remove(key);
          count++;
  
          scheduleKeyRepairsOrFinish();
      }
  
      private void complete(PaxosCleanupResponse response)
      {
          uncommittedIter.close();
          trySuccess(response);
      }
  
      private void onKeyFailure(String reason)
      {
          // not synchronized to avoid deadlock with callback we register on start
          inflight.values().forEach(AbstractPaxosRepair::cancel);
          fail(reason);
      }
  
      private synchronized void fail(String reason)
      {
          logger.info("Failing paxos cleanup session {} for {} on ranges {}. Reason: {}", session, table, ranges, reason);
          complete(PaxosCleanupResponse.failed(session, reason));
      }
  
      private void finish()
      {
          if (autoRepair)
          {
              if (count > 0)
                  logger.info("Completed {} uncommitted paxos instances for {} for session {}", count, table, session);
              logger.debug("Completed {} uncommitted paxos instances for {} on ranges {} for session {}", count, table, ranges, session);
          }
          else
              logger.info("Completed {} uncommitted paxos instances for {} on ranges {} for session {}", count, table, ranges, session);
          complete(PaxosCleanupResponse.success(session));
      }
  }
  