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.sidecar.routes.tokenrange;
  
  import java.math.BigInteger;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  
  import com.google.common.collect.Range;
  import org.junit.jupiter.api.Tag;
  import org.junit.jupiter.api.extension.ExtendWith;
  
  import io.vertx.junit5.VertxExtension;
  import io.vertx.junit5.VertxTestContext;
  import net.bytebuddy.ByteBuddy;
  import net.bytebuddy.description.type.TypeDescription;
  import net.bytebuddy.dynamic.ClassFileLocator;
  import net.bytebuddy.dynamic.TypeResolutionStrategy;
  import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
  import net.bytebuddy.implementation.MethodDelegation;
  import net.bytebuddy.implementation.bind.annotation.SuperCall;
  import net.bytebuddy.pool.TypePool;
  import org.apache.cassandra.distributed.api.IInstance;
  import org.apache.cassandra.testing.CassandraIntegrationTest;
  import org.apache.cassandra.testing.ConfigurableCassandraTestContext;
  import org.apache.cassandra.testing.IClusterExtension;
  
  import static net.bytebuddy.matcher.ElementMatchers.named;
  
  /**
   * Multi-DC Node movement scenarios integration tests for token range replica mapping endpoint with the in-jvm
   * dtest framework.
   */
  @Tag("heavy")
  @ExtendWith(VertxExtension.class)
  class MovingMultiDCTest extends MovingBaseTest
  {
      @CassandraIntegrationTest(nodesPerDc = 5, numDcs = 2, network = true, buildCluster = false)
      void retrieveMappingWhileMovingNodeMultiDC(VertxTestContext context,
                                                 ConfigurableCassandraTestContext cassandraTestContext) throws Exception
      {
          BBHelperMovingNodeMultiDC.reset();
          IClusterExtension<? extends IInstance> cluster = getMultiDCCluster(BBHelperMovingNodeMultiDC::install, cassandraTestContext);
  
          long moveTarget = getMoveTargetToken(cluster);
          runMovingTestScenario(context,
                                BBHelperMovingNodeMultiDC.transientStateStart,
                                BBHelperMovingNodeMultiDC.transientStateEnd,
                                cluster,
                                generateExpectedRangeMappingMovingNodeMultiDC(moveTarget),
                                moveTarget);
      }
  
      /**
       * Generates expected token range and replica mappings specific to the test case involving a 10 node cluster
       * across 2 DCs with the last node being moved by assigning it a different token
       * <p>
       * Expected ranges are generated by adding RF replicas per range in increasing order. The replica-sets in subsequent
       * ranges cascade with the next range excluding the first replica, and including the next replica from the nodes.
       * eg.
       * Range 1 - A, B, C
       * Range 2 - B, C, D
       * <p>
       * In this test case, the moved node is inserted between nodes 1 and 2, resulting in splitting the ranges.
       */
      private Map<String, Map<Range<BigInteger>, List<String>>>
      generateExpectedRangeMappingMovingNodeMultiDC(long moveTarget)
      {
          CassandraIntegrationTest annotation = sidecarTestContext.cassandraTestContext().annotation;
          List<Range<BigInteger>> expectedRanges = getMovingNodesExpectedRanges(annotation.nodesPerDc(),
                                                                                annotation.numDcs(),
                                                                                moveTarget);
          /*
           * The following expected ranges are generated based on the following token assignments and pending ranges.
           *
           * Token Allocations:
           * MIN TOKEN: -9223372036854775808
           * /127.0.0.1:[-5534023222112865487]
           * /127.0.0.2:[-5534023222112865486]
           * /127.0.0.3:[-1844674407370955167]
           * /127.0.0.10:[-3689348814741910327] (New Location)
           * /127.0.0.4:[-1844674407370955166]
           * /127.0.0.5:[1844674407370955153]
           * /127.0.0.6:[1844674407370955154]
           * /127.0.0.7:[5534023222112865473]
           * /127.0.0.8:[5534023222112865474]
           * /127.0.0.9:[9223372036854775793]
           * /127.0.0.10:[9223372036854775793] (Old Location)
           * MAX TOKEN: 9223372036854775807
           *
           * Pending Ranges:
           * [-5534023222112865487, -5534023222112865486]=[127.0.0.10]
           * [-5534023222112865486, -3689348814741910327]=[127.0.0.10]
           * [-1844674407370955166, 1844674407370955153]=[127.0.0.2]
           * [1844674407370955153, 1844674407370955154]=[127.0.0.2]
           * [9223372036854775794, -5534023222112865487]=[127.0.0.10]
           */
  
          Map<Range<BigInteger>, List<String>> dc1Mapping = new HashMap<>();
          Map<Range<BigInteger>, List<String>> dc2Mapping = new HashMap<>();
          // Replica 2
          dc1Mapping.put(expectedRanges.get(0), Arrays.asList("127.0.0.1", "127.0.0.3", "127.0.0.5"));
          dc2Mapping.put(expectedRanges.get(0), Arrays.asList("127.0.0.2", "127.0.0.4", "127.0.0.6",
                                                              "127.0.0.10"));
  
          dc1Mapping.put(expectedRanges.get(1), Arrays.asList("127.0.0.3", "127.0.0.5", "127.0.0.7"));
          dc2Mapping.put(expectedRanges.get(1), Arrays.asList("127.0.0.2", "127.0.0.4", "127.0.0.6",
                                                              "127.0.0.10"));
          // Split range resulting from the new token. Part 1 including the new token.
          dc1Mapping.put(expectedRanges.get(2), Arrays.asList("127.0.0.3", "127.0.0.5", "127.0.0.7"));
          dc2Mapping.put(expectedRanges.get(2), Arrays.asList("127.0.0.4", "127.0.0.6", "127.0.0.8",
                                                              "127.0.0.10"));
          // Split range resulting from the new token. Part 2 excluding new token (but starting from it)
          dc1Mapping.put(expectedRanges.get(3), Arrays.asList("127.0.0.3", "127.0.0.5", "127.0.0.7"));
          dc2Mapping.put(expectedRanges.get(3), Arrays.asList("127.0.0.4", "127.0.0.6", "127.0.0.8"));
  
          dc1Mapping.put(expectedRanges.get(4), Arrays.asList("127.0.0.5", "127.0.0.7", "127.0.0.9"));
          dc2Mapping.put(expectedRanges.get(4), Arrays.asList("127.0.0.4", "127.0.0.6", "127.0.0.8"));
  
          dc1Mapping.put(expectedRanges.get(5), Arrays.asList("127.0.0.5", "127.0.0.7", "127.0.0.9"));
          dc2Mapping.put(expectedRanges.get(5), Arrays.asList("127.0.0.6", "127.0.0.8", "127.0.0.10",
                                                              "127.0.0.2"));
  
          dc1Mapping.put(expectedRanges.get(6), Arrays.asList("127.0.0.7", "127.0.0.9", "127.0.0.1"));
          dc2Mapping.put(expectedRanges.get(6), Arrays.asList("127.0.0.6", "127.0.0.8", "127.0.0.10",
                                                              "127.0.0.2"));
  
          dc1Mapping.put(expectedRanges.get(7), Arrays.asList("127.0.0.7", "127.0.0.9", "127.0.0.1"));
          dc2Mapping.put(expectedRanges.get(7), Arrays.asList("127.0.0.8", "127.0.0.10", "127.0.0.2"));
  
          dc1Mapping.put(expectedRanges.get(8), Arrays.asList("127.0.0.9", "127.0.0.1", "127.0.0.3"));
          dc2Mapping.put(expectedRanges.get(8), Arrays.asList("127.0.0.8", "127.0.0.10", "127.0.0.2"));
  
          dc1Mapping.put(expectedRanges.get(9), Arrays.asList("127.0.0.9", "127.0.0.1", "127.0.0.3"));
          dc2Mapping.put(expectedRanges.get(9), Arrays.asList("127.0.0.10", "127.0.0.2", "127.0.0.4"));
  
          dc1Mapping.put(expectedRanges.get(10), Arrays.asList("127.0.0.1", "127.0.0.3", "127.0.0.5"));
          dc2Mapping.put(expectedRanges.get(10), Arrays.asList("127.0.0.10", "127.0.0.2", "127.0.0.4"));
          // Replica 3
          dc1Mapping.put(expectedRanges.get(11), Arrays.asList("127.0.0.1", "127.0.0.3", "127.0.0.5"));
          dc2Mapping.put(expectedRanges.get(11), Arrays.asList("127.0.0.2", "127.0.0.4", "127.0.0.6",
                                                               "127.0.0.10"));
  
          return new HashMap<String, Map<Range<BigInteger>, List<String>>>()
          {
              {
                  put("datacenter1", dc1Mapping);
                  put("datacenter2", dc2Mapping);
              }
          };
      }
  
      /**
       * ByteBuddy Helper for a multiDC moving node
       */
      public static class BBHelperMovingNodeMultiDC
      {
          static CountDownLatch transientStateStart = new CountDownLatch(1);
          static CountDownLatch transientStateEnd = new CountDownLatch(1);
  
          public static void install(ClassLoader cl, Integer nodeNumber)
          {
              // Moving the 5th node in the test case
              if (nodeNumber == MULTIDC_MOVING_NODE_IDX)
              {
                  TypePool typePool = TypePool.Default.of(cl);
                  TypeDescription description = typePool.describe("org.apache.cassandra.service.RangeRelocator")
                                                        .resolve();
                  new ByteBuddy().rebase(description, ClassFileLocator.ForClassLoader.of(cl))
                                 .method(named("stream"))
                                 .intercept(MethodDelegation.to(BBHelperMovingNodeMultiDC.class))
                                 // Defer class loading until all dependencies are loaded
                                 .make(TypeResolutionStrategy.Lazy.INSTANCE, typePool)
                                 .load(cl, ClassLoadingStrategy.Default.INJECTION);
              }
          }
  
          @SuppressWarnings("unused")
          public static Future<?> stream(@SuperCall Callable<Future<?>> orig) throws Exception
          {
              Future<?> res = orig.call();
              transientStateStart.countDown();
              awaitLatchOrTimeout(transientStateEnd, 2, TimeUnit.MINUTES, "transientStateEnd");
              return res;
          }
  
          public static void reset()
          {
              transientStateStart = new CountDownLatch(1);
              transientStateEnd = new CountDownLatch(1);
          }
      }
  }
  