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.distributed.test.thresholds;
  
  import java.io.IOException;
  import java.net.InetAddress;
  import java.net.InetSocketAddress;
  import java.net.UnknownHostException;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.concurrent.CompletableFuture;
  import java.util.function.Consumer;
  
  import com.google.common.collect.ImmutableSet;
  import com.google.common.collect.Iterables;
  import org.junit.AfterClass;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.BeforeClass;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.datastax.driver.core.ResultSet;
  import com.datastax.driver.core.SimpleStatement;
  import net.bytebuddy.ByteBuddy;
  import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
  import net.bytebuddy.implementation.MethodDelegation;
  import net.bytebuddy.implementation.bind.annotation.SuperCall;
  import net.bytebuddy.implementation.bind.annotation.This;
  import org.apache.cassandra.concurrent.SEPExecutor;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.cql3.QueryProcessor;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.api.ConsistencyLevel;
  import org.apache.cassandra.distributed.api.Feature;
  import org.apache.cassandra.distributed.api.ICluster;
  import org.apache.cassandra.distributed.api.IInvokableInstance;
  import org.apache.cassandra.distributed.api.SimpleQueryResult;
  import org.apache.cassandra.distributed.test.JavaDriverUtils;
  import org.apache.cassandra.distributed.test.TestBaseImpl;
  import org.apache.cassandra.exceptions.ReadFailureException;
  import org.apache.cassandra.exceptions.RequestFailure;
  import org.apache.cassandra.exceptions.RequestFailureReason;
  import org.apache.cassandra.exceptions.TombstoneAbortException;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.service.ClientWarn;
  import org.apache.cassandra.service.QueryState;
  import org.apache.cassandra.service.reads.ReadCallback;
  import org.apache.cassandra.service.reads.thresholds.CoordinatorWarnings;
  import org.apache.cassandra.utils.Shared;
  import org.assertj.core.api.Assertions;
  import org.assertj.core.api.Condition;
  
  import static net.bytebuddy.matcher.ElementMatchers.named;
  import static org.assertj.core.api.Assertions.assertThat;
  
  public class TombstoneCountWarningTest extends TestBaseImpl
  {
      private static final Logger logger = LoggerFactory.getLogger(TombstoneCountWarningTest.class);
  
      private static final int TOMBSTONE_WARN = 50;
      private static final int TOMBSTONE_FAIL = 100;
      private static ICluster<IInvokableInstance> CLUSTER;
      private static com.datastax.driver.core.Cluster JAVA_DRIVER;
      private static com.datastax.driver.core.Session JAVA_DRIVER_SESSION;
  
      @BeforeClass
      public static void setupClass() throws IOException
      {
          logger.info("[test step : @BeforeClass] setupClass");
          Cluster.Builder builder = Cluster.build(3);
          builder.withConfig(c -> c.set("tombstone_warn_threshold", TOMBSTONE_WARN)
                                   .set("tombstone_failure_threshold", TOMBSTONE_FAIL)
                                   .with(Feature.NATIVE_PROTOCOL, Feature.GOSSIP));
          builder.withInstanceInitializer(BB::install);
          CLUSTER = builder.start();
          JAVA_DRIVER = JavaDriverUtils.create(CLUSTER);
          JAVA_DRIVER_SESSION = JAVA_DRIVER.connect();
      }
  
      @AfterClass
      public static void teardown()
      {
          logger.info("[test step : @AfterClass] teardown");
          if (JAVA_DRIVER_SESSION != null)
              JAVA_DRIVER_SESSION.close();
          if (JAVA_DRIVER != null)
              JAVA_DRIVER.close();
      }
  
      @Before
      public void setup()
      {
          logger.info("[test step : @Before] setup");
          CLUSTER.schemaChange("DROP KEYSPACE IF EXISTS " + KEYSPACE);
          init(CLUSTER);
          CLUSTER.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
      }
  
      private static void enable(boolean value)
      {
          CLUSTER.stream().forEach(i -> i.runOnInstance(() -> DatabaseDescriptor.setReadThresholdsEnabled(value)));
      }
  
      @Test
      public void noWarningsSinglePartition()
      {
          logger.info("[test step : @Test] noWarningsSinglePartition");
          noWarnings("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk=1");
      }
  
      @Test
      public void noWarningsScan()
      {
          logger.info("[test step : @Test] noWarningsScan");
          noWarnings("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk=1");
      }
  
      public void noWarnings(String cql)
      {
          Consumer<List<String>> test = warnings ->
                                        Assert.assertEquals(Collections.emptyList(), warnings);
  
          for (int i=0; i<TOMBSTONE_WARN; i++)
              CLUSTER.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, null)", ConsistencyLevel.ALL, i);
  
          for (boolean b : Arrays.asList(true, false))
          {
              enable(b);
  
              SimpleQueryResult result = CLUSTER.coordinator(1).executeWithResult(cql, ConsistencyLevel.ALL);
              test.accept(result.warnings());
              test.accept(driverQueryAll(cql).getExecutionInfo().getWarnings());
  
              assertWarnAborts(0, 0, 0);
          }
      }
  
      @Test
      public void warnThresholdSinglePartition()
      {
          logger.info("[test step : @Test] warnThresholdSinglePartition");
          warnThreshold("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", false);
      }
  
      @Test
      public void warnThresholdScan()
      {
          logger.info("[test step : @Test] warnThresholdScan");
          warnThreshold("SELECT * FROM " + KEYSPACE + ".tbl", true);
      }
  
      private void warnThreshold(String cql, boolean isScan)
      {
          for (int i = 0; i < TOMBSTONE_WARN + 1; i++)
              CLUSTER.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, null)", ConsistencyLevel.ALL, i);
  
          enable(true);
          Consumer<List<String>> testEnabled = warnings ->
                                               Assertions.assertThat(Iterables.getOnlyElement(warnings))
                                                         .contains("nodes scanned up to " + (TOMBSTONE_WARN + 1) + " tombstones and issued tombstone warnings for query " + cql);
  
          SimpleQueryResult result = CLUSTER.coordinator(1).executeWithResult(cql, ConsistencyLevel.ALL);
          testEnabled.accept(result.warnings());
          assertWarnAborts(1, 0, 0);
          testEnabled.accept(driverQueryAll(cql).getExecutionInfo().getWarnings());
          assertWarnAborts(2, 0, 0);
  
          enable(false);
          Consumer<List<String>> testDisabled = warnings -> {
              // client warnings are currently coordinator only, so if present only 1 is expected
              if (isScan)
              {
                  // Scans perform multiple ReadCommands, which will not propgate the warnings to the top-level coordinator; so no warnings are expected
                  Assertions.assertThat(warnings).isEmpty();
              }
              else
              {
                  Assertions.assertThat(Iterables.getOnlyElement(warnings))
                            .startsWith("Read " + (TOMBSTONE_WARN + 1) + " live rows and " + (TOMBSTONE_WARN + 1) + " tombstone cells for query " + cql);
              }
          };
          result = CLUSTER.coordinator(1).executeWithResult(cql, ConsistencyLevel.ALL);
          testDisabled.accept(result.warnings());
          assertWarnAborts(2, 0, 0);
          testDisabled.accept(driverQueryAll(cql).getExecutionInfo().getWarnings());
          assertWarnAborts(2, 0, 0);
      }
  
      @Test
      public void failThresholdSinglePartition() throws UnknownHostException
      {
          logger.info("[test step : @Test] failThresholdSinglePartition");
          failThreshold("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", false);
      }
  
      @Test
      public void failThresholdScan() throws UnknownHostException
      {
          logger.info("[test step : @Test] failThresholdScan");
          failThreshold("SELECT * FROM " + KEYSPACE + ".tbl", true);
      }
  
      private void failThreshold(String cql, boolean isScan) throws UnknownHostException
      {
          for (int i = 0; i < TOMBSTONE_FAIL + 1; i++)
              CLUSTER.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, ?, null)", ConsistencyLevel.ALL, i);
  
          enable(true);
          List<String> warnings = CLUSTER.get(1).callsOnInstance(() -> {
              ClientWarn.instance.captureWarnings();
              CoordinatorWarnings.init();
              try
              {
                  QueryProcessor.execute(cql, org.apache.cassandra.db.ConsistencyLevel.ALL, QueryState.forInternalCalls());
                  Assert.fail("Expected query failure");
              }
              catch (TombstoneAbortException e)
              {
                  Assert.assertTrue(e.nodes >= 1 && e.nodes <= 3);
                  Assert.assertEquals(TOMBSTONE_FAIL + 1, e.tombstones);
                  // expected, client transport returns an error message and includes client warnings
              }
              CoordinatorWarnings.done();
              CoordinatorWarnings.reset();
              return ClientWarn.instance.getWarnings();
          }).call();
          Assertions.assertThat(Iterables.getOnlyElement(warnings))
                    .contains("nodes scanned over " + (TOMBSTONE_FAIL + 1) + " tombstones and aborted the query " + cql);
  
          assertWarnAborts(0, 1, 1);
  
          try
          {
              driverQueryAll(cql);
              Assert.fail("Query should have thrown ReadFailureException");
          }
          catch (com.datastax.driver.core.exceptions.ReadFailureException e)
          {
              // without changing the client can't produce a better message...
              // client does NOT include the message sent from the server in the exception; so the message doesn't work
              // well in this case
              Assertions.assertThat(e.getMessage()).contains("responses were required but only 0 replica responded"); // can't include ', 3 failed)' as some times its 2
  
              ImmutableSet<InetAddress> expectedKeys = ImmutableSet.of(InetAddress.getByAddress(new byte[]{ 127, 0, 0, 1 }), InetAddress.getByAddress(new byte[]{ 127, 0, 0, 2 }), InetAddress.getByAddress(new byte[]{ 127, 0, 0, 3 }));
              assertThat(e.getFailuresMap())
              .hasSizeBetween(1, 3)
              // coordinator changes from run to run, so can't assert map as the key is dynamic... so assert the domain of keys and the single value expect
              .containsValue(RequestFailureReason.READ_TOO_MANY_TOMBSTONES.code)
              .hasKeySatisfying(new Condition<InetAddress>() {
                  public boolean matches(InetAddress value)
                  {
                      return expectedKeys.contains(value);
                  }
              });
          }
  
          assertWarnAborts(0, 2, 1);
  
          // when disabled warnings only happen if on the coordinator, and coordinator may not be the one replying
          // to every query
          enable(false);
          State.blockFor(CLUSTER.get(1).config().broadcastAddress());
          warnings = CLUSTER.get(1).callsOnInstance(() -> {
              ClientWarn.instance.captureWarnings();
              try
              {
                  QueryProcessor.execute(cql, org.apache.cassandra.db.ConsistencyLevel.ALL, QueryState.forInternalCalls());
                  Assert.fail("Expected query failure");
              }
              catch (ReadFailureException e)
              {
                  Assertions.assertThat(e).isNotInstanceOf(TombstoneAbortException.class);
                  Assertions.assertThat(e.failureReasonByEndpoint).isNotEmpty();
                  Assertions.assertThat(e.failureReasonByEndpoint.values())
                            .as("Non READ_TOO_MANY_TOMBSTONES exists")
                            .allMatch(RequestFailureReason.READ_TOO_MANY_TOMBSTONES::equals);
              }
              logger.warn("Checking warnings...");
              return ClientWarn.instance.getWarnings();
          }).call();
          // client warnings are currently coordinator only, so if present only 1 is expected
          if (isScan)
          {
              // Scans perform multiple ReadCommands, which will not propgate the warnings to the top-level coordinator; so no warnings are expected
              Assertions.assertThat(warnings).isNull();
          }
          else
          {
              Assertions.assertThat(Iterables.getOnlyElement(warnings))
                        .startsWith("Read " + TOMBSTONE_FAIL + " live rows and " + (TOMBSTONE_FAIL + 1) + " tombstone cells for query " + cql);
          }
  
          assertWarnAborts(0, 2, 0);
  
          State.blockFor(CLUSTER.get(1).config().broadcastAddress());
          try
          {
              driverQueryAll(cql);
              Assert.fail("Query should have thrown ReadFailureException");
          }
          catch (com.datastax.driver.core.exceptions.ReadFailureException e)
          {
              // not checking the message as different cases exist for the failure, so the fact that this failed is enough
  
              Assertions.assertThat(e.getFailuresMap())
                        .isNotEmpty();
              Assertions.assertThat(e.getFailuresMap().values())
                        .as("Non READ_TOO_MANY_TOMBSTONES exists")
                        .allMatch(i -> i.equals(RequestFailureReason.READ_TOO_MANY_TOMBSTONES.code));
          }
  
          assertWarnAborts(0, 2, 0);
      }
  
      private static long GLOBAL_READ_ABORTS = 0;
      private static void assertWarnAborts(int warns, int aborts, int globalAborts)
      {
          Assertions.assertThat(totalWarnings()).as("warnings").isEqualTo(warns);
          Assertions.assertThat(totalAborts()).as("aborts").isEqualTo(aborts);
          long expectedGlobalAborts = GLOBAL_READ_ABORTS + globalAborts;
          Assertions.assertThat(totalReadAborts()).as("global aborts").isEqualTo(expectedGlobalAborts);
          GLOBAL_READ_ABORTS = expectedGlobalAborts;
      }
  
      private static long totalWarnings()
      {
          return CLUSTER.stream().mapToLong(i -> i.metrics().getCounter("org.apache.cassandra.metrics.keyspace.ClientTombstoneWarnings." + KEYSPACE)).sum();
      }
  
      private static long totalAborts()
      {
          return CLUSTER.stream().mapToLong(i -> i.metrics().getCounter("org.apache.cassandra.metrics.keyspace.ClientTombstoneAborts." + KEYSPACE)).sum();
      }
  
      private static long totalReadAborts()
      {
          return CLUSTER.stream().mapToLong(i ->
                                            i.metrics().getCounter("org.apache.cassandra.metrics.ClientRequest.Aborts.Read-ALL")
                                            + i.metrics().getCounter("org.apache.cassandra.metrics.ClientRequest.Aborts.RangeSlice")
          ).sum();
      }
  
      private static ResultSet driverQueryAll(String cql)
      {
          return JAVA_DRIVER_SESSION.execute(new SimpleStatement(cql).setConsistencyLevel(com.datastax.driver.core.ConsistencyLevel.ALL));
      }
  
      @Shared
      public static class State
      {
          // use InetSocketAddress as InetAddressAndPort is @Isolated which means equality doesn't work due to different
          // ClassLoaders; InetSocketAddress is @Shared so safe to use between app and cluster class loaders
          public static volatile InetSocketAddress blockFor = null;
          public static volatile CompletableFuture<Void> promise = null;
  
          // called on main thread
          public static void blockFor(InetSocketAddress address)
          {
              blockFor = address;
              promise = new CompletableFuture<>();
          }
  
          // called in C* threads; non-test threads
          public static void onFailure(InetSocketAddress address)
          {
              if (address.equals(blockFor))
                  promise.complete(null);
          }
  
          // called on main thread
          public static void syncAndClear()
          {
              if (blockFor != null)
              {
                  promise.join();
                  blockFor = null;
                  promise = null;
              }
          }
      }
  
      public static class BB
      {
          private static void install(ClassLoader cl, int instanceId)
          {
              if (instanceId != 1)
                  return;
              new ByteBuddy().rebase(ReadCallback.class)
                             .method(named("awaitResults"))
                             .intercept(MethodDelegation.to(BB.class))
                             .method(named("onFailure"))
                             .intercept(MethodDelegation.to(BB.class))
                             .make()
                             .load(cl, ClassLoadingStrategy.Default.INJECTION);
              new ByteBuddy().rebase(SEPExecutor.class)
                             .method(named("maybeExecuteImmediately"))
                             .intercept(MethodDelegation.to(BB.class))
                             .make()
                             .load(cl, ClassLoadingStrategy.Default.INJECTION);
          }
  
          @SuppressWarnings("unused")
          public static void awaitResults(@SuperCall Runnable zuper)
          {
              State.syncAndClear();
              zuper.run();
          }
  
          @SuppressWarnings("unused")
          public static void onFailure(InetAddressAndPort from, RequestFailure failure, @SuperCall Runnable zuper) throws Exception
          {
              State.onFailure(new InetSocketAddress(from.getAddress(), from.getPort()));
              zuper.run();
          }
  
          // make sure to schedule the task rather than running inline...
          // this is imporant as the read may block on the local version which can get the test to include it rather than
          // block waiting, so by scheduling we make sure its always fair
          @SuppressWarnings("unused")
          public static void maybeExecuteImmediately(Runnable task, @This SEPExecutor executor)
          {
              executor.execute(task);
          }
      }
  }
  