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;
  
  import java.io.IOException;
  
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.SinglePartitionReadCommand;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.net.IVerbHandler;
  import org.apache.cassandra.net.Message;
  import org.apache.cassandra.net.MessagingService;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.KeyMigrationState;
  import org.apache.cassandra.service.paxos.Commit.Agreed;
  import org.apache.cassandra.service.paxos.PaxosPrepare.Rejected;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.tcm.ClusterMetadataService;
  import org.apache.cassandra.tracing.Tracing;
  
  import static org.apache.cassandra.exceptions.RequestFailureReason.UNKNOWN;
  import static org.apache.cassandra.net.Verb.PAXOS2_COMMIT_AND_PREPARE_REQ;
  import static org.apache.cassandra.service.consensus.migration.ConsensusKeyMigrationState.getKeyMigrationState;
  import static org.apache.cassandra.service.paxos.Paxos.newBallot;
  import static org.apache.cassandra.service.paxos.PaxosPrepare.start;
  
  public class PaxosCommitAndPrepare
  {
      public static final RequestSerializer requestSerializer = new RequestSerializer();
      public static final RequestHandler requestHandler = new RequestHandler();
  
      static PaxosPrepare commitAndPrepare(Agreed commit, Paxos.Participants participants, SinglePartitionReadCommand readCommand, boolean isWrite, boolean acceptEarlyReadSuccess)
      {
          Ballot ballot = newBallot(commit.ballot, participants.consistencyForConsensus);
          Request request = new Request(commit, ballot, participants.electorate, readCommand, isWrite, true);
          PaxosPrepare prepare = new PaxosPrepare(participants, request, acceptEarlyReadSuccess, null);
  
          Tracing.trace("Committing {}; Preparing {}", commit.ballot, ballot);
          Message<Request> message = Message.out(PAXOS2_COMMIT_AND_PREPARE_REQ, request, participants.isUrgent());
  
          start(prepare, participants, message, RequestHandler::execute);
          return prepare;
      }
  
      private static class Request extends PaxosPrepare.AbstractRequest<Request>
      {
          final Agreed commit;
  
          Request(Agreed commit, Ballot ballot, Paxos.Electorate electorate, SinglePartitionReadCommand read, boolean isWrite, boolean isForRecovery)
          {
              super(ballot, electorate, read, isWrite, isForRecovery);
              this.commit = commit;
          }
  
          private Request(Agreed commit, Ballot ballot, Paxos.Electorate electorate, DecoratedKey partitionKey, TableMetadata table, boolean isWrite, boolean isForRecovery)
          {
              super(ballot, electorate, partitionKey, table, isWrite, isForRecovery);
              this.commit = commit;
          }
  
          Request withoutRead()
          {
              return new Request(commit, ballot, electorate, partitionKey, table, isForWrite, isForRecovery);
          }
  
          public String toString()
          {
              return commit.toString("CommitAndPrepare(") + ", " + Ballot.toString(ballot) + ')';
          }
      }
  
      public static class RequestSerializer extends PaxosPrepare.AbstractRequestSerializer<Request, Agreed>
      {
          Request construct(Agreed param, Ballot ballot, Paxos.Electorate electorate, SinglePartitionReadCommand read, boolean isWrite, boolean isForRecovery)
          {
              return new Request(param, ballot, electorate, read, isWrite, isForRecovery);
          }
  
          Request construct(Agreed param, Ballot ballot, Paxos.Electorate electorate, DecoratedKey partitionKey, TableMetadata table, boolean isWrite, boolean isForRecovery)
          {
              return new Request(param, ballot, electorate, partitionKey, table, isWrite, isForRecovery);
          }
  
          @Override
          public void serialize(Request request, DataOutputPlus out, int version) throws IOException
          {
              Agreed.serializer.serialize(request.commit, out, version);
              super.serialize(request, out, version);
          }
  
          @Override
          public Request deserialize(DataInputPlus in, int version) throws IOException
          {
              Agreed committed = Agreed.serializer.deserialize(in, version);
              return deserialize(committed, in, version);
          }
  
          @Override
          public long serializedSize(Request request, int version)
          {
              return Agreed.serializer.serializedSize(request.commit, version)
                      + super.serializedSize(request, version);
          }
      }
  
      public static class RequestHandler implements IVerbHandler<Request>
      {
          @Override
          public void doVerb(Message<Request> message)
          {
              ClusterMetadataService.instance().fetchLogFromPeerOrCMS(ClusterMetadata.current(), message.from(), message.epoch());
  
              PaxosPrepare.Response response = execute(message.payload, message.from());
              if (response == null)
                  MessagingService.instance().respondWithFailure(UNKNOWN, message);
              else
                  MessagingService.instance().respond(response, message);
          }
  
          private static PaxosPrepare.Response execute(Request request, InetAddressAndPort from)
          {
              Agreed commit = request.commit;
              if (!Paxos.isInRangeAndShouldProcess(from, commit.update.partitionKey(), commit.update.metadata(), request.read != null))
                  return null;
  
              // This can be done outside the lock
              ClusterMetadata cm = ClusterMetadata.current();
              KeyMigrationState keyMigrationState = getKeyMigrationState(cm, commit.update.metadata().id, commit.update.partitionKey());
              // Make sure the operation is safe and there is no Accord state that needs application
              // Also need to know max HLC in order to accept this ballot
              long maxHLC = keyMigrationState.maybePerformAccordToPaxosKeyMigration(true);
              if (maxHLC >= commit.ballot.unixMicros())
                  return new Rejected(Ballot.atUnixMicrosWithLsb(maxHLC + 1, 0, commit.ballot.flag()));
  
              try (PaxosState state = PaxosState.get(commit))
              {
                  state.commit(commit);
                  return PaxosPrepare.RequestHandler.execute(request, state, cm);
              }
          }
      }
  }
  