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.db.compaction.unified;
  
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import com.google.common.collect.ImmutableList;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.BeforeClass;
  import org.junit.Test;
  
  import org.apache.cassandra.Util;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DiskBoundaries;
  import org.apache.cassandra.db.PartitionPosition;
  import org.apache.cassandra.db.compaction.UnifiedCompactionStrategy;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.schema.TableMetadata;
  import org.apache.cassandra.tcm.Epoch;
  import org.apache.cassandra.utils.FBUtilities;
  import org.apache.cassandra.utils.Overlaps;
  import org.mockito.Mock;
  import org.mockito.MockitoAnnotations;
  
  import static java.lang.String.format;
  import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
  import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
  import static org.junit.Assert.assertArrayEquals;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  import static org.mockito.Mockito.when;
  
  public class ControllerTest
  {
      static final double epsilon = 0.00000001;
      static final boolean allowOverlaps = false;
      static final long checkFrequency= 600L;
  
      @Mock
      ColumnFamilyStore cfs;
  
      @Mock
      TableMetadata metadata;
  
      @Mock
      UnifiedCompactionStrategy strategy;
  
      protected String keyspaceName = "TestKeyspace";
      protected DiskBoundaries diskBoundaries = new DiskBoundaries(cfs, null, null, Epoch.FIRST, 0);
  
      @BeforeClass
      public static void setUpClass()
      {
          DatabaseDescriptor.daemonInitialization();
      }
  
      @Before
      public void setUp()
      {
          MockitoAnnotations.initMocks(this);
  
          when(strategy.getMetadata()).thenReturn(metadata);
          when(strategy.getEstimatedRemainingTasks()).thenReturn(0);
  
          when(metadata.toString()).thenReturn("");
          when(cfs.getKeyspaceName()).thenAnswer(invocation -> keyspaceName);
          when(cfs.getDiskBoundaries()).thenAnswer(invocation -> diskBoundaries);
      }
  
      Controller testFromOptions(Map<String, String> options)
      {
          addOptions(false, options);
          Controller.validateOptions(options);
  
          Controller controller = Controller.fromOptions(cfs, options);
          assertNotNull(controller);
          assertNotNull(controller.toString());
  
          for (int i = 0; i < 5; i++) // simulate 5 levels
              assertEquals(Controller.DEFAULT_SURVIVAL_FACTOR, controller.getSurvivalFactor(i), epsilon);
          assertEquals(1, controller.getNumShards(0));
          assertEquals(4, controller.getNumShards(16 * 100 << 20));
          assertEquals(Overlaps.InclusionMethod.SINGLE, controller.overlapInclusionMethod());
  
          return controller;
      }
  
      @Test
      public void testValidateOptions()
      {
          testValidateOptions(false);
      }
  
      @Test
      public void testValidateOptionsIntegers()
      {
          testValidateOptions(true);
      }
  
      public void targetSSTableSizeValidator(String inputSize) 
      {
          Map<String, String> options = new HashMap<>();
          options.putIfAbsent(Controller.TARGET_SSTABLE_SIZE_OPTION, inputSize);
          assertThatExceptionOfType(ConfigurationException.class)
          .describedAs("Should have thrown a ConfigurationException when target_sstable_size is greater than Long.MAX_VALUE")
          .isThrownBy(() -> Controller.validateOptions(options))
          .withMessageContaining(format("target_sstable_size %s is out of range of Long.", inputSize));
      }
  
      @Test
      public void testCassandra20398Values()
      {
          //TARGET_SSTABLE_SIZE_OPTION = 12E899, the value reported in CASSANDRA-20398
          String inputSize = "12E899 B";
          targetSSTableSizeValidator(inputSize); 
      }
  
      @Test
      public void testValidateOptionsTargetSSTableSizeGTLongMax()
      {
          //TARGET_SSTABLE_SIZE_OPTION > LONG.MAX_VALUE 
          // the inputSize is Long.MAX_VALUE + 100
          String inputSize = "9223372036854775907 B"; 
          targetSSTableSizeValidator(inputSize); 
      }
  
      @Test
      public void testValidateOptionsTargetSSTableSizeLTMinTargetSize()
      {
          // TARGET_SSTABLE_SIZE_OPTION < Default MIN_TARGET_SSTABLE_SIZE (1048576) 
          Map<String, String> options = new HashMap<>();
          String inputSize = "1048000 B";
          options.putIfAbsent(Controller.TARGET_SSTABLE_SIZE_OPTION, inputSize);
          assertThatExceptionOfType(ConfigurationException.class)
          .describedAs("Should have thrown a ConfigurationException when target_sstable_size is less than default MIN_TARGET_SSTABLE_SIZE")
          .isThrownBy(() -> Controller.validateOptions(options))
          .withMessageContaining(format("target_sstable_size %s is not acceptable, size must be at least %s", inputSize, FBUtilities.prettyPrintMemory(Controller.MIN_TARGET_SSTABLE_SIZE)));
      }
  
      @Test
      public void testValidateOptionsTargetSSTableSizeGTIntMax()
      {
          //TEST 4: Verifying if TARGET_SSTABLE_SIZE_OPTION (3650722199) < MIN_TARGET_SSTABLE_SIZE (2581450423)
          // Previously, TARGET_SSTABLE_SIZE_OPTION * 0.7 was stored as Integer which would 3650722199 * 0.7 = 2147483647
          // By storing it in a Long, 3650722199 * 0.7 = 2581450424. If TARGET_SSTABLE_SIZE_OPTION * 0.7 is truncated, 
          //this test case will fail
          try
          {
              Map<String, String> options = new HashMap<>();
              options.putIfAbsent(Controller.TARGET_SSTABLE_SIZE_OPTION, "3650722199 B");
              options.putIfAbsent(Controller.MIN_SSTABLE_SIZE_OPTION, "2581450423 B");
              Controller.validateOptions(options);
          } catch(ConfigurationException e) {
              fail("3650722199 * 0.7 got truncated. " + e.getMessage());
          }
      }
  
      void testValidateOptions(boolean useIntegers)
      {
          Map<String, String> options = new HashMap<>();
          addOptions(useIntegers, options);
          options = Controller.validateOptions(options);
          assertTrue(options.toString(), options.isEmpty());
      }
  
      private static void addOptions(boolean useIntegers, Map<String, String> options)
      {
          String wStr = Arrays.stream(Ws)
                              .mapToObj(useIntegers ? Integer::toString : UnifiedCompactionStrategy::printScalingParameter)
                              .collect(Collectors.joining(","));
          options.putIfAbsent(Controller.SCALING_PARAMETERS_OPTION, wStr);
  
          options.putIfAbsent(Controller.ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_OPTION, Boolean.toString(allowOverlaps));
          options.putIfAbsent(Controller.EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_OPTION, Long.toString(checkFrequency));
  
          options.putIfAbsent(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(2));
          options.putIfAbsent(Controller.TARGET_SSTABLE_SIZE_OPTION, FBUtilities.prettyPrintMemory(100 << 20));
          // The below value is based on the value in the above statement. Decreasing the above statement should result in a decrease below.
          options.putIfAbsent(Controller.MIN_SSTABLE_SIZE_OPTION, "70.710MiB");
          options.putIfAbsent(Controller.OVERLAP_INCLUSION_METHOD_OPTION, toLowerCaseLocalized(Overlaps.InclusionMethod.SINGLE.toString()));
          options.putIfAbsent(Controller.SSTABLE_GROWTH_OPTION, "0.5");
      }
  
      @Test
      public void testScalingParameterConversion()
      {
          testScalingParameterConversion("T4", 2);
          testScalingParameterConversion("L4", -2);
          testScalingParameterConversion("N", 0);
          testScalingParameterConversion("L2, T2, N", 0, 0, 0);
          testScalingParameterConversion("T10, T8, T4, N, L4, L6", 8, 6, 2, 0, -2, -4);
          testScalingParameterConversion("T10000, T1000, T100, T10, T2, L10, L100, L1000, L10000", 9998, 998, 98, 8, 0, -8, -98, -998, -9998);
  
          testScalingParameterParsing("-50 ,  T5  ,  3 ,  N  ,  L7 ,  +5 , -12  ,T9,L4,6,-7,+0,-0", -50, 3, 3, 0, -5, 5, -12, 7, -2, 6, -7, 0, 0);
  
          testScalingParameterError("Q6");
          testScalingParameterError("L4,,T5");
          testScalingParameterError("L1");
          testScalingParameterError("T1");
          testScalingParameterError("L0");
          testScalingParameterError("T0");
          testScalingParameterError("T-5");
          testScalingParameterError("T+5");
          testScalingParameterError("L-5");
          testScalingParameterError("L+5");
          testScalingParameterError("N3");
          testScalingParameterError("7T");
          testScalingParameterError("T,5");
          testScalingParameterError("L,5");
      }
  
      void testScalingParameterConversion(String definition, int... parameters)
      {
          testScalingParameterParsing(definition, parameters);
  
          String normalized = definition.replaceAll("T2|L2", "N");
          assertEquals(normalized, Controller.printScalingParameters(parameters));
  
          testScalingParameterParsing(Arrays.toString(parameters).replaceAll("[\\[\\]]", ""), parameters);
      }
  
      void testScalingParameterParsing(String definition, int... parameters)
      {
          assertArrayEquals(parameters, Controller.parseScalingParameters(definition));
      }
  
      void testScalingParameterError(String definition)
      {
          try
          {
              Controller.parseScalingParameters(definition);
              Assert.fail("Expected error on " + definition);
          }
          catch (ConfigurationException e)
          {
              // expected
          }
      }
  
      @Test
      public void testGetNumShards()
      {
          Map<String, String> options = new HashMap<>();
          options.putIfAbsent(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.putIfAbsent(Controller.TARGET_SSTABLE_SIZE_OPTION, FBUtilities.prettyPrintMemory(100 << 20));
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "0B");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.0");
          Controller.validateOptions(options);
          Controller controller = Controller.fromOptions(cfs, options);
  
          // Easy ones
          // x00 MiB = x * 100
          assertEquals(6, controller.getNumShards(Math.scalb(600, 20)));
          assertEquals(24, controller.getNumShards(Math.scalb(2400, 20)));
          assertEquals(6 * 1024, controller.getNumShards(Math.scalb(600, 30)));
          // Check rounding
          assertEquals(6, controller.getNumShards(Math.scalb(800, 20)));
          assertEquals(12, controller.getNumShards(Math.scalb(900, 20)));
          assertEquals(6 * 1024, controller.getNumShards(Math.scalb(800, 30)));
          assertEquals(12 * 1024, controller.getNumShards(Math.scalb(900, 30)));
          // Check lower limit
          assertEquals(3, controller.getNumShards(Math.scalb(200, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(100, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(10, 20)));
          assertEquals(3, controller.getNumShards(5));
          assertEquals(3, controller.getNumShards(0));
          // Check upper limit
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(600, 40)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(10, 60)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Double.POSITIVE_INFINITY));
      }
  
      @Test
      public void testGetNumShards_growth_0()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.0");
          Controller controller = Controller.fromOptions(cfs, options);
          assertEquals(0.0, controller.sstableGrowthModifier, 0.0);
  
          // Easy ones
          // x00 MiB = x * 100
          assertEquals(6, controller.getNumShards(Math.scalb(600, 20)));
          assertEquals(24, controller.getNumShards(Math.scalb(2400, 20)));
          assertEquals(6 * 1024, controller.getNumShards(Math.scalb(600, 30)));
          // Check rounding
          assertEquals(6, controller.getNumShards(Math.scalb(800, 20)));
          assertEquals(12, controller.getNumShards(Math.scalb(900, 20)));
          assertEquals(6 * 1024, controller.getNumShards(Math.scalb(800, 30)));
          assertEquals(12 * 1024, controller.getNumShards(Math.scalb(900, 30)));
          // Check lower limit
          assertEquals(3, controller.getNumShards(Math.scalb(200, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(100, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(50, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(30, 20)));
          // Check min size
          assertEquals(1, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
          // Check upper limit
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(600, 40)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(10, 60)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Double.POSITIVE_INFINITY));
          // Check NaN
          assertEquals(1, controller.getNumShards(Double.NaN));
      }
  
      @Test
      public void testGetNumShards_growth_1()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "1.0");
          Controller controller = Controller.fromOptions(cfs, options);
  
          // Easy ones
          // x00 MiB = x * 100
          assertEquals(3, controller.getNumShards(Math.scalb(600, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(2400, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(600, 30)));
          // Check rounding
          assertEquals(3, controller.getNumShards(Math.scalb(800, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(900, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(800, 30)));
          assertEquals(3, controller.getNumShards(Math.scalb(900, 30)));
          // Check lower limit
          assertEquals(3, controller.getNumShards(Math.scalb(200, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(100, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(50, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(30, 20)));
          // Check min size
          assertEquals(1, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
          // Check upper limit
          assertEquals(3, controller.getNumShards(Math.scalb(600, 40)));
          assertEquals(3, controller.getNumShards(Math.scalb(10, 60)));
          assertEquals(3, controller.getNumShards(Double.POSITIVE_INFINITY));
          // Check NaN
          assertEquals(1, controller.getNumShards(Double.NaN));
      }
  
      @Test
      public void testGetNumShards_growth_1_2()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.5");
          Controller controller = Controller.fromOptions(cfs, options);
  
          // Easy ones
          // x00 MiB = x * 3 * 100
          assertEquals(3 * 2, controller.getNumShards(Math.scalb(4 * 3 * 100, 20)));
          assertEquals(3 * 4, controller.getNumShards(Math.scalb(16 * 3 * 100, 20)));
          assertEquals(3 * 32, controller.getNumShards(Math.scalb(3 * 100, 20 + 10)));
          // Check rounding. Note: Size must grow by 2x to get sqrt(2) times more shards.
          assertEquals(6, controller.getNumShards(Math.scalb(2350, 20)));
          assertEquals(12, controller.getNumShards(Math.scalb(2450, 20)));
          assertEquals(3 * 32, controller.getNumShards(Math.scalb(550, 30)));
          assertEquals(6 * 32, controller.getNumShards(Math.scalb(650, 30)));
          // Check lower limit
          assertEquals(3, controller.getNumShards(Math.scalb(200, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(100, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(50, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(30, 20)));
          // Check min size
          assertEquals(1, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
          // Check upper limit
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(600, 60)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(10, 80)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Double.POSITIVE_INFINITY));
          // Check NaN
          assertEquals(1, controller.getNumShards(Double.NaN));
      }
  
      @Test
      public void testGetNumShards_growth_1_3()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.333");
          Controller controller = Controller.fromOptions(cfs, options);
  
          // Easy ones
          // x00 MiB = x * 3 * 100
          assertEquals(3 * 4, controller.getNumShards(Math.scalb(8 * 3 * 100, 20)));
          assertEquals(3 * 16, controller.getNumShards(Math.scalb(64 * 3 * 100, 20)));
          assertEquals(3 * 64, controller.getNumShards(Math.scalb(3 * 100, 20 + 9)));
          // Check rounding. Note: size must grow by 2 ^ 3/4 to get sqrt(2) times more shards
          assertEquals(12, controller.getNumShards(Math.scalb(4000, 20)));
          assertEquals(24, controller.getNumShards(Math.scalb(4100, 20)));
          assertEquals(3 * 64, controller.getNumShards(Math.scalb(500, 29)));
          assertEquals(6 * 64, controller.getNumShards(Math.scalb(550, 29)));
          // Check lower limit
          assertEquals(3, controller.getNumShards(Math.scalb(200, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(100, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(50, 20)));
          assertEquals(3, controller.getNumShards(Math.scalb(30, 20)));
          // Check min size
          assertEquals(1, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
          // Check upper limit
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(600, 50)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Math.scalb(10, 60)));
          assertEquals(3 * (int) Controller.MAX_SHARD_SPLIT, controller.getNumShards(Double.POSITIVE_INFINITY));
          assertEquals(1, controller.getNumShards(Double.NaN));
      }
  
      @Test
      public void testGetNumShards_minSize_10MiB_b_3()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(3));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.333");
          Controller controller = Controller.fromOptions(cfs, options);
          // Check min size
          assertEquals(1, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
      }
  
      @Test
      public void testGetNumShards_minSize_10MiB_b_20()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(20));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.333");
          Controller controller = Controller.fromOptions(cfs, options);
          // Check min size
          assertEquals(2, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(2, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
      }
  
      @Test
      public void testGetNumShards_minSize_10MiB_b_8()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(8));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "10MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.333");
          Controller controller = Controller.fromOptions(cfs, options);
          // Check min size
          assertEquals(2, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(2, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(1, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
      }
  
      @Test
      public void testGetNumShards_minSize_3MiB_b_20()
      {
          Map<String, String> options = new HashMap<>();
          options.put(Controller.BASE_SHARD_COUNT_OPTION, Integer.toString(20));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, "100MiB");
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "3MiB");
          options.put(Controller.SSTABLE_GROWTH_OPTION, "0.333");
          Controller controller = Controller.fromOptions(cfs, options);
          // Check min size
          assertEquals(4, controller.getNumShards(Math.scalb(29, 20)));
          assertEquals(4, controller.getNumShards(Math.scalb(20, 20)));
          assertEquals(4, controller.getNumShards(Math.scalb(19, 20)));
          assertEquals(1, controller.getNumShards(5));
          assertEquals(1, controller.getNumShards(0));
      }
  
      static final int[] Ws = new int[] { 30, 2, 0, -6};
  
      @Test
      public void testFromOptions()
      {
          Map<String, String> options = new HashMap<>();
          addOptions(false, options);
  
          Controller controller = testFromOptions(options);
  
          for (int i = 0; i < Ws.length; i++)
              assertEquals(Ws[i], controller.getScalingParameter(i));
  
          assertEquals(Ws[Ws.length-1], controller.getScalingParameter(Ws.length));
      }
  
      @Test
      public void testFromOptionsIntegers()
      {
          Map<String, String> options = new HashMap<>();
          addOptions(true, options);
  
          Controller controller = testFromOptions(options);
  
          for (int i = 0; i < Ws.length; i++)
              assertEquals(Ws[i], controller.getScalingParameter(i));
  
          assertEquals(Ws[Ws.length-1], controller.getScalingParameter(Ws.length));
      }
  
      @Test
      public void testMaxSSTablesToCompact()
      {
          Map<String, String> options = new HashMap<>();
          Controller controller = testFromOptions(options);
          assertTrue(controller.maxSSTablesToCompact == Integer.MAX_VALUE);
  
          options.put(Controller.MAX_SSTABLES_TO_COMPACT_OPTION, "100");
          controller = testFromOptions(options);
          assertEquals(100, controller.maxSSTablesToCompact);
      }
  
      @Test
      public void testExpiredSSTableCheckFrequency()
      {
          Map<String, String> options = new HashMap<>();
  
          Controller controller = testFromOptions(options);
          assertEquals(TimeUnit.MILLISECONDS.convert(Controller.DEFAULT_EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS, TimeUnit.SECONDS),
                       controller.getExpiredSSTableCheckFrequency());
  
          options.put(Controller.EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_OPTION, "5");
          controller = testFromOptions(options);
          assertEquals(5000L, controller.getExpiredSSTableCheckFrequency());
  
          try
          {
              options.put(Controller.EXPIRED_SSTABLE_CHECK_FREQUENCY_SECONDS_OPTION, "0");
              testFromOptions(options);
              fail("Exception should be thrown");
          }
          catch (ConfigurationException e)
          {
              // valid path
          }
      }
  
      @Test
      public void testAllowOverlaps()
      {
          Map<String, String> options = new HashMap<>();
  
          Controller controller = testFromOptions(options);
          assertEquals(Controller.DEFAULT_ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION, controller.getIgnoreOverlapsInExpirationCheck());
  
          options.put(Controller.ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_OPTION, "true");
          controller = testFromOptions(options);
          assertEquals(Controller.ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION, controller.getIgnoreOverlapsInExpirationCheck());
      }
  
      @Test
      public void testBaseShardCountDefault()
      {
          Map<String, String> options = new HashMap<>();
          Controller controller = Controller.fromOptions(cfs, options);
          assertEquals(Controller.DEFAULT_BASE_SHARD_COUNT, controller.baseShardCount);
  
          PartitionPosition min = Util.testPartitioner().getMinimumToken().minKeyBound();
          diskBoundaries = new DiskBoundaries(cfs, null, ImmutableList.of(min, min, min), Epoch.FIRST, 0);
          controller = Controller.fromOptions(cfs, options);
          assertEquals(4, controller.baseShardCount);
  
          diskBoundaries = new DiskBoundaries(cfs, null, ImmutableList.of(min), Epoch.FIRST, 0);
          controller = Controller.fromOptions(cfs, options);
          assertEquals(Controller.DEFAULT_BASE_SHARD_COUNT, controller.baseShardCount);
      }
  
      @Test
      public void testMinSSTableSize()
      {
          Map<String, String> options = new HashMap<>();
  
          // verify 0 is acceptable
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, format("%sB", 0));
          Controller.validateOptions(options);
  
          // test min < 0 failes
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, "-1B");
          assertThatExceptionOfType(ConfigurationException.class)
          .describedAs("Should have thrown a ConfigurationException when min_sstable_size is less than 0")
          .isThrownBy(() -> Controller.validateOptions(options))
          .withMessageContaining("greater than or equal to 0");
  
          // test min < default target sstable size * INV_SQRT_2
          int limit = (int) Math.ceil(Controller.DEFAULT_TARGET_SSTABLE_SIZE * Controller.INVERSE_SQRT_2);
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, format("%sB", limit + 1));
          assertThatExceptionOfType(ConfigurationException.class)
          .describedAs("Should have thrown a ConfigurationException when min_sstable_size is greater than target_sstable_size")
          .isThrownBy(() -> Controller.validateOptions(options))
          .withMessageContaining(format("Invalid configuration, %s (%s) should be less than 70%% of the targetSSTableSize (%s)", Controller.MIN_SSTABLE_SIZE_OPTION,  FBUtilities.prettyPrintMemory(limit+1), FBUtilities.prettyPrintMemory(Controller.DEFAULT_TARGET_SSTABLE_SIZE)));
  
          // test min < configured target table size * INV_SQRT_2
          limit = (int) Math.ceil(Controller.MIN_TARGET_SSTABLE_SIZE * 2 * Controller.INVERSE_SQRT_2);
          options.put(Controller.MIN_SSTABLE_SIZE_OPTION, format("%sB", limit + 1));
          options.put(Controller.TARGET_SSTABLE_SIZE_OPTION, format("%sB", Controller.MIN_TARGET_SSTABLE_SIZE * 2));
  
          assertThatExceptionOfType(ConfigurationException.class)
          .describedAs("Should have thrown a ConfigurationException when min_sstable_size is greater than target_sstable_size")
          .isThrownBy(() -> Controller.validateOptions(options))
          .withMessageContaining(format("Invalid configuration, %s (%s) should be less than 70%% of the targetSSTableSize (%s)", Controller.MIN_SSTABLE_SIZE_OPTION, FBUtilities.prettyPrintMemory(limit + 1), FBUtilities.prettyPrintMemory(Controller.MIN_TARGET_SSTABLE_SIZE * 2)));
      }
  }