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.tcm.transformations;
  
  import java.io.IOException;
  import java.util.HashSet;
  import java.util.Set;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.db.TypeSizes;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.tcm.ClusterMetadata;
  import org.apache.cassandra.tcm.ClusterMetadataService;
  import org.apache.cassandra.tcm.Transformation;
  import org.apache.cassandra.tcm.membership.NodeId;
  import org.apache.cassandra.tcm.membership.NodeState;
  import org.apache.cassandra.tcm.ownership.PlacementDeltas;
  import org.apache.cassandra.tcm.ownership.PlacementProvider;
  import org.apache.cassandra.tcm.ownership.PlacementTransitionPlan;
  import org.apache.cassandra.tcm.sequences.BootstrapAndReplace;
  import org.apache.cassandra.tcm.sequences.LockedRanges;
  import org.apache.cassandra.tcm.serialization.AsymmetricMetadataSerializer;
  import org.apache.cassandra.tcm.serialization.Version;
  
  import static org.apache.cassandra.exceptions.ExceptionCode.INVALID;
  
  public class PrepareReplace implements Transformation
  {
      private static final Logger logger = LoggerFactory.getLogger(PrepareReplace.class);
  
      public static final Serializer serializer = new Serializer();
  
      private final NodeId replaced;
      private final NodeId replacement;
      private final PlacementProvider placementProvider;
      private final boolean joinTokenRing;
      private final boolean streamData;
  
      public PrepareReplace(NodeId replaced,
                            NodeId replacement,
                            PlacementProvider placementProvider,
                            boolean joinTokenRing,
                            boolean streamData)
      {
          this.replaced = replaced;
          this.replacement = replacement;
          this.placementProvider = placementProvider;
          this.joinTokenRing = joinTokenRing;
          this.streamData = streamData;
      }
  
      @Override
      public Kind kind()
      {
          return Kind.PREPARE_REPLACE;
      }
  
      public NodeId replacement()
      {
          return replacement;
      }
  
      @Override
      public Result execute(ClusterMetadata prev)
      {
          if (prev.directory.peerState(replaced) != NodeState.JOINED)
              return new Rejected(INVALID, String.format("Rejecting this plan as the replaced node %s is in state %s", replaced, prev.directory.peerState(replaced)));
  
          if (prev.directory.peerState(replacement) != NodeState.REGISTERED)
              return new Rejected(INVALID, String.format("Rejecting this plan as the replacement node %s is in state %s", replacement, prev.directory.peerState(replacement)));
  
          LockedRanges.Key unlockKey = LockedRanges.keyFor(prev.nextEpoch());
          LockedRanges lockedRanges = prev.lockedRanges;
  
          PlacementTransitionPlan transitionPlan = placementProvider.planForReplacement(prev,
                                                                                        replaced,
                                                                                        replacement,
                                                                                        prev.schema.getKeyspaces());
  
          LockedRanges.AffectedRanges rangesToLock = transitionPlan.affectedRanges();
          LockedRanges.Key alreadyLockedBy = lockedRanges.intersects(rangesToLock);
  
          if (!alreadyLockedBy.equals(LockedRanges.NOT_LOCKED))
          {
              return new Rejected(INVALID, String.format("Rejecting this plan as it interacts with a range locked by %s (locked: %s, new: %s)",
                                                         alreadyLockedBy, lockedRanges, rangesToLock));
          }
  
          StartReplace start = new StartReplace(replaced, replacement, transitionPlan.addToWrites(), unlockKey);
          MidReplace mid = new MidReplace(replaced, replacement, transitionPlan.moveReads(), unlockKey);
          FinishReplace finish = new FinishReplace(replaced, replacement, transitionPlan.removeFromWrites(), unlockKey);
          transitionPlan.assertPreExistingWriteReplica(prev.placements);
  
          Set<Token> tokens = new HashSet<>(prev.tokenMap.tokens(replaced));
          BootstrapAndReplace plan = BootstrapAndReplace.newSequence(prev.nextEpoch(),
                                                                     unlockKey,
                                                                     tokens,
                                                                     start, mid, finish,
                                                                     joinTokenRing, streamData);
  
          LockedRanges newLockedRanges = lockedRanges.lock(unlockKey, rangesToLock);
          ClusterMetadata.Transformer proposed = prev.transformer()
                                                     .with(newLockedRanges)
                                                     .with(prev.inProgressSequences.with(replacement(), plan));
          logger.info("Node {} is replacing {}, tokens {}", prev.directory.endpoint(replacement), prev.directory.endpoint(replaced), prev.tokenMap.tokens(replaced));
          return Transformation.success(proposed, rangesToLock);
      }
  
      @Override
      public String toString()
      {
          return "PrepareReplace{" +
                 "replaced=" + replaced +
                 ", replacement=" + replacement +
                 ", joinTokenRing=" + joinTokenRing +
                 ", streamData=" + streamData +
                 '}';
      }
  
      public static final class Serializer implements AsymmetricMetadataSerializer<Transformation, PrepareReplace>
      {
          public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
          {
              assert t instanceof PrepareReplace;
              PrepareReplace transformation = (PrepareReplace) t;
              NodeId.serializer.serialize(transformation.replaced, out, version);
              NodeId.serializer.serialize(transformation.replacement, out, version);
              out.writeBoolean(transformation.joinTokenRing);
              out.writeBoolean(transformation.streamData);
          }
  
          public PrepareReplace deserialize(DataInputPlus in, Version version) throws IOException
          {
              NodeId replaced = NodeId.serializer.deserialize(in, version);
              NodeId replacement = NodeId.serializer.deserialize(in, version);
              boolean joinTokenRing = in.readBoolean();
              boolean streamData = in.readBoolean();
              return new PrepareReplace(replaced,
                                        replacement,
                                        ClusterMetadataService.instance().placementProvider(),
                                        joinTokenRing,
                                        streamData);
          }
  
          public long serializedSize(Transformation t, Version version)
          {
              assert t instanceof PrepareReplace;
              PrepareReplace transformation = (PrepareReplace) t;
              return NodeId.serializer.serializedSize(transformation.replaced, version) +
                     NodeId.serializer.serializedSize(transformation.replacement, version) +
                     (TypeSizes.BOOL_SIZE * 2);
          }
      }
  
      public static abstract class BaseReplaceTransformation extends ApplyPlacementDeltas
      {
          protected final NodeId replaced;
  
          public BaseReplaceTransformation(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key unlockKey, boolean unlock)
          {
              super(replacement, delta, unlockKey, unlock);
              this.replaced = replaced;
          }
  
          public NodeId replacement()
          {
              return nodeId;
          }
  
          public NodeId replaced()
          {
              return replaced;
          }
      }
  
      // Analogous to ApplyPlacementDeltas.SerializerBase. The difference is that classes which extend
      // BaseReplaceTransformation have 2 NodeIds, whereas the equivalents from PrepareJoin and PrepareLeave only have a single id
      // TODO we can probably make the hierarchy of abstract/concrete transformations and associated serializers a bit less convoluted
      public abstract static class SerializerBase<T extends BaseReplaceTransformation> implements AsymmetricMetadataSerializer<Transformation, T>
      {
          public void serialize(Transformation t, DataOutputPlus out, Version version) throws IOException
          {
              BaseReplaceTransformation change = (T) t;
              NodeId.serializer.serialize(change.replaced(), out, version);
              NodeId.serializer.serialize(change.replacement(), out, version);
              PlacementDeltas.serializer.serialize(change.delta, out, version);
              LockedRanges.Key.serializer.serialize(change.lockKey, out, version);
          }
  
          public T deserialize(DataInputPlus in, Version version) throws IOException
          {
              NodeId replaced = NodeId.serializer.deserialize(in, version);
              NodeId replacement = NodeId.serializer.deserialize(in, version);
              PlacementDeltas delta = PlacementDeltas.serializer.deserialize(in, version);
              LockedRanges.Key lockKey = LockedRanges.Key.serializer.deserialize(in, version);
              return construct(replaced, replacement, delta, lockKey);
          }
  
          public long serializedSize(Transformation t, Version version)
          {
              BaseReplaceTransformation change = (T) t;
  
              return NodeId.serializer.serializedSize(change.replaced(), version) +
                     NodeId.serializer.serializedSize(change.replacement(), version) +
                     PlacementDeltas.serializer.serializedSize(change.delta, version) +
                     LockedRanges.Key.serializer.serializedSize(change.lockKey, version);
          }
  
          abstract T construct(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key lockKey);
      }
  
      //  This is functionally identical to StartJoin. They only exist as distinct transformations for clarity.
      public static class StartReplace extends BaseReplaceTransformation
      {
          public static final Serializer serializer = new Serializer();
  
          public StartReplace(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key unlockKey)
          {
              super(replaced, replacement, delta, unlockKey, false);
          }
  
          @Override
          public Kind kind()
          {
              return Kind.START_REPLACE;
          }
  
          @Override
          public ClusterMetadata.Transformer transform(ClusterMetadata prev, ClusterMetadata.Transformer transformer)
          {
              return transformer.withNodeState(replacement(), NodeState.BOOT_REPLACING)
                                .with(prev.inProgressSequences.with(nodeId, (BootstrapAndReplace plan) -> plan.advance(prev.nextEpoch())));
          }
  
          @Override
          public String toString()
          {
              return "StartReplace{" +
                     ", delta=" + delta +
                     ", lockKey=" + lockKey +
                     ", unlock=" + unlock +
                     ", replaced=" + replaced +
                     ", replacement=" + replacement() +
                     '}';
          }
  
          public static final class Serializer extends PrepareReplace.SerializerBase<StartReplace>
          {
              @Override
              public StartReplace construct(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key unlockKey)
              {
                  return new StartReplace(replaced, replacement, delta, unlockKey);
              }
          }
      }
  
      public static class MidReplace extends BaseReplaceTransformation
      {
          public static final Serializer serializer = new Serializer();
  
          public MidReplace(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key unlockKey)
          {
              super(replaced, replacement, delta, unlockKey, false);
          }
  
          @Override
          public Kind kind()
          {
              return Kind.MID_REPLACE;
          }
  
          @Override
          public ClusterMetadata.Transformer transform(ClusterMetadata prev, ClusterMetadata.Transformer transformer)
          {
              return transformer
                     .with(prev.inProgressSequences.with(nodeId, (plan) -> plan.advance(prev.nextEpoch())));
          }
  
          @Override
          public String toString()
          {
              return "MidReplace{" +
                     ", delta=" + delta +
                     ", lockKey=" + lockKey +
                     ", unlock=" + unlock +
                     ", replaced=" + replaced +
                     ", replacement=" + replacement() +
                     '}';
          }
  
          public static final class Serializer extends PrepareReplace.SerializerBase<MidReplace>
          {
              @Override
              MidReplace construct(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key lockKey)
              {
                  return new MidReplace(replaced, replacement, delta, lockKey);
              }
          }
      }
  
      public static class FinishReplace extends BaseReplaceTransformation
      {
          public static final Serializer serializer = new Serializer();
  
          public FinishReplace(NodeId replaced,
                               NodeId replacement,
                               PlacementDeltas delta,
                               LockedRanges.Key unlockKey)
          {
              super(replaced, replacement, delta, unlockKey, true);
          }
  
          @Override
          public Kind kind()
          {
              return Kind.FINISH_REPLACE;
          }
  
          @Override
          public ClusterMetadata.Transformer transform(ClusterMetadata prev, ClusterMetadata.Transformer transformer)
          {
              return transformer.replaced(replaced, replacement())
                                .with(prev.inProgressSequences.without(nodeId));
          }
  
          @Override
          public String toString()
          {
              return "FinishReplace{" +
                     "replaced=" + replaced +
                     ", replacement=" + replacement() +
                     ", delta=" + delta +
                     ", unlockKey=" + lockKey +
                     '}';
          }
  
          public static final class Serializer extends PrepareReplace.SerializerBase<FinishReplace>
          {
              @Override
              FinishReplace construct(NodeId replaced, NodeId replacement, PlacementDeltas delta, LockedRanges.Key lockKey)
              {
                  return new FinishReplace(replaced, replacement, delta, lockKey);
              }
          }
      }
  }