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.io.sstable;
  
  import java.io.BufferedWriter;
  import java.io.IOException;
  import java.net.UnknownHostException;
  import java.nio.channels.FileChannel;
  import java.nio.file.Files;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.zip.CRC32;
  import java.util.zip.CheckedInputStream;
  
  import com.google.common.base.Charsets;
  import org.apache.commons.lang3.StringUtils;
  import org.junit.Assume;
  import org.junit.BeforeClass;
  import org.junit.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.UpdateBuilder;
  import org.apache.cassandra.Util;
  import org.apache.cassandra.batchlog.Batch;
  import org.apache.cassandra.batchlog.BatchlogManager;
  import org.apache.cassandra.cache.ChunkCache;
  import org.apache.cassandra.config.DatabaseDescriptor;
  import org.apache.cassandra.cql3.QueryProcessor;
  import org.apache.cassandra.db.BufferDecoratedKey;
  import org.apache.cassandra.db.ColumnFamilyStore;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.Keyspace;
  import org.apache.cassandra.db.PartitionPosition;
  import org.apache.cassandra.db.compaction.CompactionManager;
  import org.apache.cassandra.db.marshal.UUIDType;
  import org.apache.cassandra.dht.ByteOrderedPartitioner.BytesToken;
  import org.apache.cassandra.dht.Murmur3Partitioner;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.exceptions.WriteTimeoutException;
  import org.apache.cassandra.io.FSWriteError;
  import org.apache.cassandra.io.sstable.format.SSTableReader;
  import org.apache.cassandra.io.sstable.format.SSTableReaderWithFilter;
  import org.apache.cassandra.io.sstable.format.SortedTableVerifier.RangeOwnHelper;
  import org.apache.cassandra.io.sstable.format.big.BigFormat;
  import org.apache.cassandra.io.sstable.format.big.BigFormat.Components;
  import org.apache.cassandra.io.sstable.format.bti.BtiFormat;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.io.util.FileInputStreamPlus;
  import org.apache.cassandra.io.util.FileUtils;
  import org.apache.cassandra.io.util.RandomAccessReader;
  import org.apache.cassandra.locator.InetAddressAndPort;
  import org.apache.cassandra.schema.CompressionParams;
  import org.apache.cassandra.schema.KeyspaceParams;
  import org.apache.cassandra.service.reads.range.TokenUpdater;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.OutputHandler;
  
  import static org.apache.cassandra.SchemaLoader.counterCFMD;
  import static org.apache.cassandra.SchemaLoader.createKeyspace;
  import static org.apache.cassandra.SchemaLoader.loadSchema;
  import static org.apache.cassandra.SchemaLoader.standardCFMD;
  import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  
  /**
   * Test for {@link IVerifier}.
   * <p>
   * Note: the complete coverage is composed of:
   * - {@link org.apache.cassandra.tools.StandaloneVerifierOnSSTablesTest}
   * - {@link org.apache.cassandra.tools.StandaloneVerifierTest}
   * - {@link VerifyTest}
   */
  public class VerifyTest
  {
      private final static Logger logger = LoggerFactory.getLogger(VerifyTest.class);
  
      public static final String KEYSPACE = "Keyspace1";
      public static final String CF = "Standard1";
      public static final String CF2 = "Standard2";
      public static final String CF3 = "Standard3";
      public static final String CF4 = "Standard4";
      public static final String COUNTER_CF = "Counter1";
      public static final String COUNTER_CF2 = "Counter2";
      public static final String COUNTER_CF3 = "Counter3";
      public static final String COUNTER_CF4 = "Counter4";
      public static final String CORRUPT_CF = "Corrupt1";
      public static final String CORRUPT_CF2 = "Corrupt2";
      public static final String CORRUPT_CF3 = "Corrupt3";
      public static final String CORRUPTCOUNTER_CF = "CounterCorrupt1";
      public static final String CORRUPTCOUNTER_CF2 = "CounterCorrupt2";
  
      public static final String CF_UUID = "UUIDKeys";
      public static final String BF_ALWAYS_PRESENT = "BfAlwaysPresent";
  
      @BeforeClass
      public static void defineSchema() throws ConfigurationException
      {
          CompressionParams compressionParameters = CompressionParams.snappy(32768);
          DatabaseDescriptor.daemonInitialization();
          DatabaseDescriptor.setColumnIndexSizeInKiB(0);
  
          loadSchema();
          createKeyspace(KEYSPACE,
                         KeyspaceParams.simple(1),
                         standardCFMD(KEYSPACE, CF).compression(compressionParameters),
                         standardCFMD(KEYSPACE, CF2).compression(compressionParameters),
                         standardCFMD(KEYSPACE, CF3),
                         standardCFMD(KEYSPACE, CF4),
                         standardCFMD(KEYSPACE, CORRUPT_CF),
                         standardCFMD(KEYSPACE, CORRUPT_CF2),
                         standardCFMD(KEYSPACE, CORRUPT_CF3),
                         counterCFMD(KEYSPACE, COUNTER_CF).compression(compressionParameters),
                         counterCFMD(KEYSPACE, COUNTER_CF2).compression(compressionParameters),
                         counterCFMD(KEYSPACE, COUNTER_CF3),
                         counterCFMD(KEYSPACE, COUNTER_CF4),
                         counterCFMD(KEYSPACE, CORRUPTCOUNTER_CF),
                         counterCFMD(KEYSPACE, CORRUPTCOUNTER_CF2),
                         standardCFMD(KEYSPACE, CF_UUID, 0, UUIDType.instance),
                         standardCFMD(KEYSPACE, BF_ALWAYS_PRESENT).bloomFilterFpChance(1.0));
      }
  
  
      @Test
      public void testVerifyCorrect()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testVerifyCounterCorrect()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
  
          fillCounterCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testExtendedVerifyCorrect()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF2);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testExtendedVerifyCounterCorrect()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF2);
  
          fillCounterCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).extendedVerification(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testVerifyCorrectUncompressed()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF3);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testVerifyCounterCorrectUncompressed()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF3);
  
          fillCounterCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testExtendedVerifyCorrectUncompressed()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF4);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().extendedVerification(true).invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
      @Test
      public void testExtendedVerifyCounterCorrectUncompressed()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF4);
  
          fillCounterCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().extendedVerification(true).invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
          }
          catch (CorruptSSTableException err)
          {
              fail("Unexpected CorruptSSTableException");
          }
      }
  
  
      @Test
      public void testVerifyIncorrectDigest() throws IOException, WriteTimeoutException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF);
  
          fillCF(cfs, 2);
  
          Util.getAll(Util.cmd(cfs).build());
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
  
          try (RandomAccessReader file = RandomAccessReader.open(sstable.descriptor.fileFor(Components.DIGEST)))
          {
              long correctChecksum = file.readLong();
  
              writeChecksum(++correctChecksum, sstable.descriptor.fileFor(Components.DIGEST));
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a CorruptSSTableException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(false).build()))
          {
              verifier.verify();
              fail("Expected a RuntimeException to be thrown");
          }
          catch (RuntimeException expected)
          {
          }
      }
  
  
      @Test
      public void testVerifyCorruptRowCorrectDigest() throws IOException, WriteTimeoutException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF2);
  
          fillCF(cfs, 2);
  
          Util.getAll(Util.cmd(cfs).build());
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          // overwrite one row with garbage
          long row0Start = sstable.getPosition(PartitionPosition.ForKey.get(ByteBufferUtil.bytes("0"), cfs.getPartitioner()), SSTableReader.Operator.EQ);
          long row1Start = sstable.getPosition(PartitionPosition.ForKey.get(ByteBufferUtil.bytes("1"), cfs.getPartitioner()), SSTableReader.Operator.EQ);
          long startPosition = Math.min(row0Start, row1Start);
          long endPosition = Math.max(row0Start, row1Start);
  
          try (FileChannel file = new File(sstable.getFilename()).newReadWriteChannel()) {
              file.position(startPosition);
              file.write(ByteBufferUtil.bytes(StringUtils.repeat('z', 2)));
          }
          if (ChunkCache.instance != null)
              ChunkCache.instance.invalidateFile(sstable.getFilename());
  
          // Update the Digest to have the right Checksum
          writeChecksum(simpleFullChecksum(sstable.getFilename()), sstable.descriptor.fileFor(Components.DIGEST));
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              // First a simple verify checking digest, which should succeed
              try
              {
                  verifier.verify();
              }
              catch (CorruptSSTableException err)
              {
                  logger.error("Unexpected exception", err);
                  fail("Simple verify should have succeeded as digest matched");
              }
          }
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).extendedVerification(true).build()))
          {
              // Now try extended verify
              try
              {
                  verifier.verify();
              }
              catch (CorruptSSTableException err)
              {
                  return;
              }
              fail("Expected a CorruptSSTableException to be thrown");
          }
      }
  
      @Test
      public void testVerifyBrokenSSTableMetadata() throws IOException, WriteTimeoutException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF2);
          cfs.truncateBlocking();
          fillCF(cfs, 2);
  
          Util.getAll(Util.cmd(cfs).build());
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
          FileChannel file = sstable.descriptor.fileFor(Components.STATS).newReadWriteChannel();
          file.position(0);
          file.write(ByteBufferUtil.bytes(StringUtils.repeat('z', 2)));
          file.close();
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a CorruptSSTableException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(false).build()))
          {
              verifier.verify();
              fail("Expected a RuntimeException to be thrown");
          }
          catch (CorruptSSTableException unexpected)
          {
              fail("wrong exception thrown");
          }
          catch (RuntimeException expected)
          {
          }
      }
  
      @Test
      public void testVerifyMutateRepairStatus() throws IOException, WriteTimeoutException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF2);
          cfs.truncateBlocking();
          fillCF(cfs, 2);
  
          Util.getAll(Util.cmd(cfs).build());
  
          // make the sstable repaired:
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          sstable.descriptor.getMetadataSerializer().mutateRepairMetadata(sstable.descriptor, System.currentTimeMillis(), sstable.getPendingRepair(), sstable.isTransient());
          sstable.reloadSSTableMetadata();
  
          // break the sstable:
          long correctChecksum;
          try (RandomAccessReader file = RandomAccessReader.open(sstable.descriptor.fileFor(Components.DIGEST)))
          {
              correctChecksum = file.readLong();
          }
          writeChecksum(++correctChecksum, sstable.descriptor.fileFor(Components.DIGEST));
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().mutateRepairStatus(false).invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a CorruptSSTableException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
  
          assertTrue(sstable.isRepaired());
  
          // now the repair status should be changed:
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().mutateRepairStatus(true).invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a CorruptSSTableException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
          assertFalse(sstable.isRepaired());
      }
  
      @Test(expected = RuntimeException.class)
      public void testOutOfRangeTokens() throws IOException
      {
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
          fillCF(cfs, 100);
          byte[] tk1 = new byte[1], tk2 = new byte[1];
          tk1[0] = 2;
          tk2[0] = 1;
          new TokenUpdater().withTokens(InetAddressAndPort.getByName("127.0.0.1"), new BytesToken(tk1))
                            .withTokens(InetAddressAndPort.getByName("127.0.0.2"), new BytesToken(tk2))
                            .update();
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().checkOwnsTokens(true).extendedVerification(true).build()))
          {
              verifier.verify();
          }
      }
  
      @Test
      public void testMutateRepair() throws IOException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF3);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          sstable.descriptor.getMetadataSerializer().mutateRepairMetadata(sstable.descriptor, 1, sstable.getPendingRepair(), sstable.isTransient());
          sstable.reloadSSTableMetadata();
          cfs.getTracker().notifySSTableRepairedStatusChanged(Collections.singleton(sstable));
          assertTrue(sstable.isRepaired());
          cfs.forceMajorCompaction();
  
          sstable = cfs.getLiveSSTables().iterator().next();
          long correctChecksum;
          try (RandomAccessReader file = RandomAccessReader.open(sstable.descriptor.fileFor(Components.DIGEST)))
          {
              correctChecksum = file.readLong();
          }
          writeChecksum(++correctChecksum, sstable.descriptor.fileFor(Components.DIGEST));
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).mutateRepairStatus(true).build()))
          {
              verifier.verify();
              fail("should be corrupt");
          }
          catch (CorruptSSTableException expected)
          {
          }
          assertFalse(sstable.isRepaired());
      }
  
      @Test
      public void testVerifyIndex() throws IOException
      {
          if (BigFormat.isSelected())
              testBrokenComponentHelper(BigFormat.Components.PRIMARY_INDEX);
          else if (BtiFormat.isSelected())
              testBrokenComponentHelper(BtiFormat.Components.PARTITION_INDEX);
          else
              throw Util.testMustBeImplementedForSSTableFormat();
      }
  
      @Test
      public void testVerifyBf() throws IOException
      {
          Assume.assumeTrue(SSTableReaderWithFilter.class.isAssignableFrom(DatabaseDescriptor.getSelectedSSTableFormat().getReaderFactory().getReaderClass()));
          testBrokenComponentHelper(Components.FILTER);
      }
  
      @Test
      public void testVerifyIndexSummary() throws IOException
      {
          Assume.assumeTrue(BigFormat.isSelected());
          testBrokenComponentHelper(Components.SUMMARY);
      }
  
      private void testBrokenComponentHelper(Component componentToBreak) throws IOException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF2);
  
          fillCF(cfs, 2);
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().build()))
          {
              verifier.verify(); //still not corrupt, should pass
          }
          try (FileChannel fileChannel = sstable.descriptor.fileFor(componentToBreak).newReadWriteChannel())
          {
              fileChannel.truncate(3);
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("should throw exception");
          }
          catch (CorruptSSTableException e)
          {
              //expected
          }
      }
  
      @Test
      public void testQuick() throws IOException
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CORRUPT_CF);
  
          fillCF(cfs, 2);
  
          Util.getAll(Util.cmd(cfs).build());
  
          SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
  
  
          try (RandomAccessReader file = RandomAccessReader.open(sstable.descriptor.fileFor(Components.DIGEST)))
          {
              long correctChecksum = file.readLong();
  
              writeChecksum(++correctChecksum, sstable.descriptor.fileFor(Components.DIGEST));
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a CorruptSSTableException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).quick(true).build())) // with quick = true we don't verify the digest
          {
              verifier.verify();
          }
  
          try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().invokeDiskFailurePolicy(true).build()))
          {
              verifier.verify();
              fail("Expected a RuntimeException to be thrown");
          }
          catch (CorruptSSTableException expected)
          {
          }
      }
  
      @Test
      public void testRangeOwnHelper()
      {
          List<Range<Token>> normalized = new ArrayList<>();
          normalized.add(r(Long.MIN_VALUE, Long.MIN_VALUE + 1));
          normalized.add(r(Long.MIN_VALUE + 5, Long.MIN_VALUE + 6));
          normalized.add(r(Long.MIN_VALUE + 10, Long.MIN_VALUE + 11));
          normalized.add(r(0, 10));
          normalized.add(r(10, 11));
          normalized.add(r(20, 25));
          normalized.add(r(26, 200));
  
          RangeOwnHelper roh = new RangeOwnHelper(normalized);
  
          roh.validate(dk(1));
          roh.validate(dk(10));
          roh.validate(dk(11));
          roh.validate(dk(21));
          roh.validate(dk(25));
          boolean gotException = false;
          try
          {
              roh.validate(dk(26));
          }
          catch (Throwable t)
          {
              gotException = true;
          }
          assertTrue(gotException);
      }
  
      @Test(expected = AssertionError.class)
      public void testRangeOwnHelperBadToken()
      {
          List<Range<Token>> normalized = new ArrayList<>();
          normalized.add(r(0, 10));
          RangeOwnHelper roh = new RangeOwnHelper(normalized);
          roh.validate(dk(1));
          // call with smaller token to get exception
          roh.validate(dk(0));
      }
  
  
      @Test
      public void testRangeOwnHelperNormalize()
      {
          List<Range<Token>> normalized = Range.normalize(Collections.singletonList(r(0, 0)));
          RangeOwnHelper roh = new RangeOwnHelper(normalized);
          roh.validate(dk(Long.MIN_VALUE));
          roh.validate(dk(0));
          roh.validate(dk(Long.MAX_VALUE));
      }
  
      @Test
      public void testRangeOwnHelperNormalizeWrap()
      {
          List<Range<Token>> normalized = Range.normalize(Collections.singletonList(r(Long.MAX_VALUE - 1000, Long.MIN_VALUE + 1000)));
          RangeOwnHelper roh = new RangeOwnHelper(normalized);
          roh.validate(dk(Long.MIN_VALUE));
          roh.validate(dk(Long.MAX_VALUE));
          boolean gotException = false;
          try
          {
              roh.validate(dk(26));
          }
          catch (Throwable t)
          {
              gotException = true;
          }
          assertTrue(gotException);
      }
  
      @Test
      public void testEmptyRanges()
      {
          new RangeOwnHelper(Collections.emptyList()).validate(dk(1));
      }
  
      @Test
      public void testVerifyLocalPartitioner() throws UnknownHostException
      {
          byte[] tk1 = new byte[1], tk2 = new byte[1];
          tk1[0] = 2;
          tk2[0] = 1;
  //        tmd.updateNormalToken(new ByteOrderedPartitioner.BytesToken(tk1), InetAddressAndPort.getByName("127.0.0.1"));
  //        tmd.updateNormalToken(new ByteOrderedPartitioner.BytesToken(tk2), InetAddressAndPort.getByName("127.0.0.2"));
          // write some bogus to a localpartitioner table
          Batch bogus = Batch.createLocal(nextTimeUUID(), 0, Collections.emptyList());
          BatchlogManager.store(bogus);
          ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("batches");
          Util.flush(cfs);
          for (SSTableReader sstable : cfs.getLiveSSTables())
          {
  
              try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().checkOwnsTokens(true).build()))
              {
                  verifier.verify();
              }
          }
      }
  
      @Test
      public void testNoFilterFile()
      {
          CompactionManager.instance.disableAutoCompaction();
          Keyspace keyspace = Keyspace.open(KEYSPACE);
          ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(BF_ALWAYS_PRESENT);
          fillCF(cfs, 100);
          assertEquals(1.0, cfs.metadata().params.bloomFilterFpChance, 0.0);
          for (SSTableReader sstable : cfs.getLiveSSTables())
          {
              File f = sstable.descriptor.fileFor(Components.FILTER);
              assertFalse(f.exists());
              try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options().build()))
              {
                  verifier.verify();
              }
          }
      }
  
      @Test
      public void testVerifyReversedPartitioner()
      {
          for (long i = 0; i < 10; i++)
              QueryProcessor.executeInternal("insert into system.local_metadata_log (epoch) values (?)", i);
          ColumnFamilyStore cfs = Keyspace.open("system").getColumnFamilyStore("local_metadata_log");
          cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
          assertFalse(cfs.getLiveSSTables().isEmpty());
          for (SSTableReader sstable : cfs.getLiveSSTables())
          {
              try (IVerifier verifier = sstable.getVerifier(cfs, new OutputHandler.LogOutput(), false, IVerifier.options()
                                                                                                                .checkOwnsTokens(true).build()))
              {
                  verifier.verify();
              }
          }
      }
  
      private DecoratedKey dk(long l)
      {
          return new BufferDecoratedKey(t(l), ByteBufferUtil.EMPTY_BYTE_BUFFER);
      }
  
      private Range<Token> r(long s, long e)
      {
          return new Range<>(t(s), t(e));
      }
  
      private Token t(long t)
      {
          return new Murmur3Partitioner.LongToken(t);
      }
  
  
      protected void fillCF(ColumnFamilyStore cfs, int partitionsPerSSTable)
      {
          for (int i = 0; i < partitionsPerSSTable; i++)
          {
              UpdateBuilder.create(cfs.metadata(), String.valueOf(i))
                           .newRow("c1").add("val", "1")
                           .newRow("c2").add("val", "2")
                           .apply();
          }
  
          Util.flush(cfs);
      }
  
      protected void fillCounterCF(ColumnFamilyStore cfs, int partitionsPerSSTable) throws WriteTimeoutException
      {
          for (int i = 0; i < partitionsPerSSTable; i++)
          {
              UpdateBuilder.create(cfs.metadata(), String.valueOf(i))
                           .newRow("c1").add("val", 100L)
                           .apply();
          }
  
          Util.flush(cfs);
      }
  
      protected long simpleFullChecksum(String filename) throws IOException
      {
          try (FileInputStreamPlus inputStream = new FileInputStreamPlus(filename);
               CheckedInputStream cinStream = new CheckedInputStream(inputStream, new CRC32()))
          {
              byte[] b = new byte[128];
              //noinspection StatementWithEmptyBody
              while (cinStream.read(b) >= 0)
              {
              }
              return cinStream.getChecksum().getValue();
          }
      }
  
      public static void writeChecksum(long checksum, File file)
      {
          BufferedWriter out = null;
          try
          {
              out = Files.newBufferedWriter(file.toPath(), Charsets.UTF_8);
              out.write(String.valueOf(checksum));
              out.flush();
              out.close();
          }
          catch (IOException e)
          {
              throw new FSWriteError(e, file);
          }
          finally
          {
              FileUtils.closeQuietly(out);
          }
      }
  }
  