cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject [05/10] cassandra git commit: Merge branch 'cassandra-2.2' into cassandra-3.0
Date Fri, 01 Apr 2016 17:47:35 GMT
Merge branch 'cassandra-2.2' into cassandra-3.0


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/15131391
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/15131391
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/15131391

Branch: refs/heads/cassandra-3.5
Commit: 15131391d833c1a0ba0e26903e27867fd101fe72
Parents: 98086b6 c662259
Author: Yuki Morishita <yukim@apache.org>
Authored: Fri Apr 1 12:44:52 2016 -0500
Committer: Yuki Morishita <yukim@apache.org>
Committed: Fri Apr 1 12:44:52 2016 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../db/lifecycle/LogAwareFileLister.java        | 23 -------
 .../apache/cassandra/io/sstable/Descriptor.java | 20 +++++--
 .../db/lifecycle/LogTransactionTest.java        | 63 +++++++++++++++-----
 .../cassandra/io/sstable/DescriptorTest.java    | 40 +++++++++----
 5 files changed, 90 insertions(+), 57 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/15131391/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 482c41a,c8a4f21..d8d4834
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,21 -1,9 +1,22 @@@
 -2.2.6
 +3.0.5
 + * Improve backoff policy for cqlsh COPY FROM (CASSANDRA-11320)
 + * Improve IF NOT EXISTS check in CREATE INDEX (CASSANDRA-11131)
 + * Upgrade ohc to 0.4.3
 + * Enable SO_REUSEADDR for JMX RMI server sockets (CASSANDRA-11093)
 + * Allocate merkletrees with the correct size (CASSANDRA-11390)
 + * Support streaming pre-3.0 sstables (CASSANDRA-10990)
 + * Add backpressure to compressed commit log (CASSANDRA-10971)
 + * SSTableExport supports secondary index tables (CASSANDRA-11330)
 + * Fix sstabledump to include missing info in debug output (CASSANDRA-11321)
 + * Establish and implement canonical bulk reading workload(s) (CASSANDRA-10331)
 + * Fix paging for IN queries on tables without clustering columns (CASSANDRA-11208)
 + * Remove recursive call from CompositesSearcher (CASSANDRA-11304)
 + * Fix filtering on non-primary key columns for queries without index (CASSANDRA-6377)
 + * Fix sstableloader fail when using materialized view (CASSANDRA-11275)
 +Merged from 2.2:
+  * Use canonical path for directory in SSTable descriptor (CASSANDRA-10587)
   * Add cassandra-stress keystore option (CASSANDRA-9325)
 - * Fix out-of-space error treatment in memtable flushing (CASSANDRA-11448).
   * Dont mark sstables as repairing with sub range repairs (CASSANDRA-11451)
 - * Fix use of NullUpdater for 2i during compaction (CASSANDRA-11450)
   * Notify when sstables change after cancelling compaction (CASSANDRA-11373)
   * cqlsh: COPY FROM should check that explicit column names are valid (CASSANDRA-11333)
   * Add -Dcassandra.start_gossip startup option (CASSANDRA-10809)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/15131391/src/java/org/apache/cassandra/db/lifecycle/LogAwareFileLister.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/lifecycle/LogAwareFileLister.java
index 01bcb8a,0000000..3393b5c
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/db/lifecycle/LogAwareFileLister.java
+++ b/src/java/org/apache/cassandra/db/lifecycle/LogAwareFileLister.java
@@@ -1,195 -1,0 +1,172 @@@
 +package org.apache.cassandra.db.lifecycle;
 +
 +import java.io.File;
 +import java.io.IOException;
 +import java.nio.file.DirectoryStream;
 +import java.nio.file.Files;
 +import java.nio.file.Path;
 +import java.util.*;
 +import java.util.function.BiFunction;
 +import java.util.stream.Collectors;
 +import java.util.stream.Stream;
 +import java.util.stream.StreamSupport;
 +
 +import com.google.common.annotations.VisibleForTesting;
 +
 +import org.apache.cassandra.db.Directories;
 +
 +import static org.apache.cassandra.db.Directories.*;
 +
 +/**
 + * A class for listing files in a folder.
 + */
 +final class LogAwareFileLister
 +{
 +    // The folder to scan
 +    private final Path folder;
 +
 +    // The filter determines which files the client wants returned
 +    private final BiFunction<File, FileType, Boolean> filter; //file, file type
 +
 +    // The behavior when we fail to list files
 +    private final OnTxnErr onTxnErr;
 +
 +    // The unfiltered result
 +    NavigableMap<File, Directories.FileType> files = new TreeMap<>();
 +
 +    @VisibleForTesting
 +    LogAwareFileLister(Path folder, BiFunction<File, FileType, Boolean> filter, OnTxnErr onTxnErr)
 +    {
 +        this.folder = folder;
 +        this.filter = filter;
 +        this.onTxnErr = onTxnErr;
 +    }
 +
 +    public List<File> list()
 +    {
 +        try
 +        {
 +            return innerList();
 +        }
 +        catch (Throwable t)
 +        {
 +            throw new RuntimeException(String.format("Failed to list files in %s", folder), t);
 +        }
 +    }
 +
 +    List<File> innerList() throws Throwable
 +    {
 +        list(Files.newDirectoryStream(folder))
 +        .stream()
 +        .filter((f) -> !LogFile.isLogFile(f))
 +        .forEach((f) -> files.put(f, FileType.FINAL));
 +
 +        // Since many file systems are not atomic, we cannot be sure we have listed a consistent disk state
 +        // (Linux would permit this, but for simplicity we keep our behaviour the same across platforms)
 +        // so we must be careful to list txn log files AFTER every other file since these files are deleted last,
 +        // after all other files are removed
 +        list(Files.newDirectoryStream(folder, '*' + LogFile.EXT))
 +        .stream()
 +        .filter(LogFile::isLogFile)
 +        .forEach(this::classifyFiles);
 +
 +        // Finally we apply the user filter before returning our result
 +        return files.entrySet().stream()
 +                    .filter((e) -> filter.apply(e.getKey(), e.getValue()))
 +                    .map(Map.Entry::getKey)
 +                    .collect(Collectors.toList());
 +    }
 +
 +    static List<File> list(DirectoryStream<Path> stream) throws IOException
 +    {
 +        try
 +        {
 +            return StreamSupport.stream(stream.spliterator(), false)
 +                                .map(Path::toFile)
 +                                .filter((f) -> !f.isDirectory())
 +                                .collect(Collectors.toList());
 +        }
 +        finally
 +        {
 +            stream.close();
 +        }
 +    }
 +
 +    /**
 +     * We read txn log files, if we fail we throw only if the user has specified
 +     * OnTxnErr.THROW, else we log an error and apply the txn log anyway
 +     */
 +    void classifyFiles(File txnFile)
 +    {
 +        LogFile txn = LogFile.make(txnFile);
 +        readTxnLog(txn);
 +        classifyFiles(txn);
 +        files.put(txnFile, FileType.TXN_LOG);
 +    }
 +
 +    void readTxnLog(LogFile txn)
 +    {
 +        if (!txn.verify() && onTxnErr == OnTxnErr.THROW)
 +            throw new LogTransaction.CorruptTransactionLogException("Some records failed verification. See earlier in log for details.", txn);
 +    }
 +
 +    void classifyFiles(LogFile txnFile)
 +    {
 +        Map<LogRecord, Set<File>> oldFiles = txnFile.getFilesOfType(files.navigableKeySet(), LogRecord.Type.REMOVE);
 +        Map<LogRecord, Set<File>> newFiles = txnFile.getFilesOfType(files.navigableKeySet(), LogRecord.Type.ADD);
 +
 +        if (txnFile.completed())
 +        { // last record present, filter regardless of disk status
 +            setTemporary(txnFile, oldFiles.values(), newFiles.values());
 +            return;
 +        }
 +
 +        if (allFilesPresent(txnFile, oldFiles, newFiles))
 +        {  // all files present, transaction is in progress, this will filter as aborted
 +            setTemporary(txnFile, oldFiles.values(), newFiles.values());
 +            return;
 +        }
 +
 +        // some files are missing, we expect the txn file to either also be missing or completed, so check
 +        // disk state again to resolve any previous races on non-atomic directory listing platforms
 +
 +        // if txn file also gone, then do nothing (all temporary should be gone, we could remove them if any)
 +        if (!txnFile.exists())
 +            return;
 +
 +        // otherwise read the file again to see if it is completed now
 +        readTxnLog(txnFile);
 +
 +        if (txnFile.completed())
 +        { // if after re-reading the txn is completed then filter accordingly
 +            setTemporary(txnFile, oldFiles.values(), newFiles.values());
 +            return;
 +        }
 +
 +        // some files are missing and yet the txn is still there and not completed
 +        // something must be wrong (see comment at the top of this file requiring txn to be
 +        // completed before obsoleting or aborting sstables)
 +        throw new RuntimeException(String.format("Failed to list directory files in %s, inconsistent disk state for transaction %s",
 +                                                 folder,
 +                                                 txnFile));
 +    }
 +
 +    /** See if all files are present or if only the last record files are missing and it's a NEW record */
 +    private static boolean allFilesPresent(LogFile txnFile, Map<LogRecord, Set<File>> oldFiles, Map<LogRecord, Set<File>> newFiles)
 +    {
 +        LogRecord lastRecord = txnFile.getLastRecord();
 +        return !Stream.concat(oldFiles.entrySet().stream(),
 +                              newFiles.entrySet().stream()
 +                                      .filter((e) -> e.getKey() != lastRecord))
 +                      .filter((e) -> e.getKey().numFiles > e.getValue().size())
 +                      .findFirst().isPresent();
 +    }
 +
 +    private void setTemporary(LogFile txnFile, Collection<Set<File>> oldFiles, Collection<Set<File>> newFiles)
 +    {
 +        Collection<Set<File>> temporary = txnFile.committed() ? oldFiles : newFiles;
 +        temporary.stream()
 +                 .flatMap(Set::stream)
 +                 .forEach((f) -> this.files.put(f, FileType.TEMPORARY));
 +    }
- 
-     @VisibleForTesting
-     static Set<File> getTemporaryFiles(File folder)
-     {
-         return listFiles(folder, FileType.TEMPORARY);
-     }
- 
-     @VisibleForTesting
-     static Set<File> getFinalFiles(File folder)
-     {
-         return listFiles(folder, FileType.FINAL);
-     }
- 
-     @VisibleForTesting
-     static Set<File> listFiles(File folder, FileType ... types)
-     {
-         Collection<FileType> match = Arrays.asList(types);
-         return new LogAwareFileLister(folder.toPath(),
-                                       (file, type) -> match.contains(type),
-                                       OnTxnErr.IGNORE).list()
-                                                       .stream()
-                                                       .collect(Collectors.toSet());
-     }
 +}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/15131391/src/java/org/apache/cassandra/io/sstable/Descriptor.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/sstable/Descriptor.java
index dbce56c,ed81616..ff4abfc
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@@ -18,10 -18,12 +18,12 @@@
  package org.apache.cassandra.io.sstable;
  
  import java.io.File;
+ import java.io.IOError;
+ import java.io.IOException;
 -import java.util.ArrayDeque;
 -import java.util.Deque;
 -import java.util.StringTokenizer;
 +import java.util.*;
 +import java.util.regex.Pattern;
  
 +import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.CharMatcher;
  import com.google.common.base.Objects;
  
@@@ -46,7 -47,20 +48,9 @@@ import static org.apache.cassandra.io.s
   */
  public class Descriptor
  {
 -
 -    public static enum Type
 -    {
 -        TEMP("tmp", true), TEMPLINK("tmplink", true), FINAL(null, false);
 -        public final boolean isTemporary;
 -        public final String marker;
 -        Type(String marker, boolean isTemporary)
 -        {
 -            this.isTemporary = isTemporary;
 -            this.marker = marker;
 -        }
 -    }
 +    public static String TMP_EXT = ".tmp";
+ 
+     /** canonicalized path to the directory where SSTable resides */
      public final File directory;
      /** version has the following format: <code>[a-z]+</code> */
      public final Version version;
@@@ -89,10 -104,10 +100,10 @@@
          this.ksname = ksname;
          this.cfname = cfname;
          this.generation = generation;
 -        this.type = temp;
          this.formatType = formatType;
 +        this.digestComponent = digestComponent;
  
-         hashCode = Objects.hashCode(version, directory, generation, ksname, cfname, formatType);
 -        hashCode = Objects.hashCode(version, this.directory, generation, ksname, cfname, temp, formatType);
++        hashCode = Objects.hashCode(version, this.directory, generation, ksname, cfname, formatType);
      }
  
      public Descriptor withGeneration(int newGeneration)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/15131391/test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java
index 4f2fc73,0000000..45b5844
mode 100644,000000..100644
--- a/test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java
+++ b/test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java
@@@ -1,1183 -1,0 +1,1214 @@@
 +/*
 + * 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.lifecycle;
 +
 +import java.io.File;
++import java.io.IOError;
 +import java.io.IOException;
 +import java.io.RandomAccessFile;
 +import java.nio.file.Files;
 +import java.util.*;
 +import java.util.function.BiConsumer;
 +import java.util.function.Consumer;
 +import java.util.stream.Collectors;
 +
 +import com.google.common.collect.ImmutableSet;
 +import com.google.common.collect.Iterables;
 +import com.google.common.collect.Sets;
 +import org.junit.BeforeClass;
 +import org.junit.Test;
 +
 +import static junit.framework.Assert.assertNotNull;
 +import static junit.framework.Assert.assertNull;
 +import static junit.framework.Assert.fail;
 +import static org.junit.Assert.assertEquals;
 +import static org.junit.Assert.assertFalse;
 +import static org.junit.Assert.assertTrue;
 +import junit.framework.Assert;
 +import org.apache.cassandra.MockSchema;
 +import org.apache.cassandra.db.ColumnFamilyStore;
 +import org.apache.cassandra.db.Directories;
 +import org.apache.cassandra.db.SerializationHeader;
 +import org.apache.cassandra.db.compaction.*;
 +import org.apache.cassandra.io.sstable.*;
 +import org.apache.cassandra.io.sstable.format.SSTableReader;
 +import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 +import org.apache.cassandra.io.sstable.metadata.MetadataType;
 +import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
 +import org.apache.cassandra.io.util.BufferedSegmentedFile;
 +import org.apache.cassandra.io.util.ChannelProxy;
 +import org.apache.cassandra.io.util.FileUtils;
 +import org.apache.cassandra.io.util.RandomAccessReader;
 +import org.apache.cassandra.io.util.SegmentedFile;
 +import org.apache.cassandra.utils.AlwaysPresentFilter;
 +import org.apache.cassandra.utils.concurrent.AbstractTransactionalTest;
 +import org.apache.cassandra.utils.concurrent.Transactional;
 +
 +public class LogTransactionTest extends AbstractTransactionalTest
 +{
 +    private static final String KEYSPACE = "TransactionLogsTest";
 +
 +    @BeforeClass
 +    public static void setUp()
 +    {
 +        MockSchema.cleanup();
 +    }
 +
 +    protected AbstractTransactionalTest.TestableTransaction newTest() throws Exception
 +    {
 +        LogTransaction.waitForDeletions();
 +        SSTableReader.resetTidying();
 +        return new TxnTest();
 +    }
 +
 +    private static final class TxnTest extends TestableTransaction
 +    {
 +        private final static class Transaction extends Transactional.AbstractTransactional implements Transactional
 +        {
 +            final ColumnFamilyStore cfs;
 +            final LogTransaction txnLogs;
 +            final File dataFolder;
 +            final SSTableReader sstableOld;
 +            final SSTableReader sstableNew;
 +            final LogTransaction.SSTableTidier tidier;
 +
 +            Transaction(ColumnFamilyStore cfs, LogTransaction txnLogs) throws IOException
 +            {
 +                this.cfs = cfs;
 +                this.txnLogs = txnLogs;
 +                this.dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +                this.sstableOld = sstable(dataFolder, cfs, 0, 128);
 +                this.sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +                assertNotNull(txnLogs);
 +                assertNotNull(txnLogs.id());
 +                Assert.assertEquals(OperationType.COMPACTION, txnLogs.type());
 +
 +                txnLogs.trackNew(sstableNew);
 +                tidier = txnLogs.obsoleted(sstableOld);
 +                assertNotNull(tidier);
 +            }
 +
 +            protected Throwable doCommit(Throwable accumulate)
 +            {
 +                sstableOld.markObsolete(tidier);
 +                sstableOld.selfRef().release();
 +                LogTransaction.waitForDeletions();
 +
 +                Throwable ret = txnLogs.commit(accumulate);
 +
 +                sstableNew.selfRef().release();
 +                return ret;
 +            }
 +
 +            protected Throwable doAbort(Throwable accumulate)
 +            {
 +                tidier.abort();
 +                LogTransaction.waitForDeletions();
 +
 +                Throwable ret = txnLogs.abort(accumulate);
 +
 +                sstableNew.selfRef().release();
 +                sstableOld.selfRef().release();
 +                return ret;
 +            }
 +
 +            protected void doPrepare()
 +            {
 +                txnLogs.prepareToCommit();
 +            }
 +
 +            void assertInProgress() throws Exception
 +            {
 +                assertFiles(dataFolder.getPath(), Sets.newHashSet(Iterables.concat(sstableNew.getAllFilePaths(),
 +                                                                                   sstableOld.getAllFilePaths(),
 +                                                                                   txnLogs.logFilePaths())));
 +            }
 +
 +            void assertPrepared() throws Exception
 +            {
 +            }
 +
 +            void assertAborted() throws Exception
 +            {
 +                assertFiles(dataFolder.getPath(), new HashSet<>(sstableOld.getAllFilePaths()));
 +            }
 +
 +            void assertCommitted() throws Exception
 +            {
 +                assertFiles(dataFolder.getPath(), new HashSet<>(sstableNew.getAllFilePaths()));
 +            }
 +        }
 +
 +        final Transaction txn;
 +
 +        private TxnTest() throws IOException
 +        {
 +            this(MockSchema.newCFS(KEYSPACE));
 +        }
 +
 +        private TxnTest(ColumnFamilyStore cfs) throws IOException
 +        {
 +            this(cfs, new LogTransaction(OperationType.COMPACTION));
 +        }
 +
 +        private TxnTest(ColumnFamilyStore cfs, LogTransaction txnLogs) throws IOException
 +        {
 +            this(new Transaction(cfs, txnLogs));
 +        }
 +
 +        private TxnTest(Transaction txn)
 +        {
 +            super(txn);
 +            this.txn = txn;
 +        }
 +
 +        protected void assertInProgress() throws Exception
 +        {
 +            txn.assertInProgress();
 +        }
 +
 +        protected void assertPrepared() throws Exception
 +        {
 +            txn.assertPrepared();
 +        }
 +
 +        protected void assertAborted() throws Exception
 +        {
 +            txn.assertAborted();
 +        }
 +
 +        protected void assertCommitted() throws Exception
 +        {
 +            txn.assertCommitted();
 +        }
 +    }
 +
 +    @Test
 +    public void testUntrack() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        // complete a transaction without keep the new files since they were untracked
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +        log.untrackNew(sstableNew);
 +
 +        log.finish();
 +
 +        sstableNew.selfRef().release();
 +        Thread.sleep(1);
 +        LogTransaction.waitForDeletions();
 +
 +        assertFiles(dataFolder.getPath(), Collections.<String>emptySet());
 +    }
 +
 +    @Test
 +    public void testCommitSameDesc() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableOld1 = sstable(dataFolder, cfs, 0, 128);
 +        SSTableReader sstableOld2 = sstable(dataFolder, cfs, 0, 256);
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +
 +        sstableOld1.setReplaced();
 +
 +        LogTransaction.SSTableTidier tidier = log.obsoleted(sstableOld2);
 +        assertNotNull(tidier);
 +
 +        log.finish();
 +
 +        sstableOld2.markObsolete(tidier);
 +
 +        sstableOld1.selfRef().release();
 +        sstableOld2.selfRef().release();
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>(sstableNew.getAllFilePaths()));
 +
 +        sstableNew.selfRef().release();
 +    }
 +
 +    @Test
 +    public void testCommitOnlyNew() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstable);
 +        log.finish();
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>(sstable.getAllFilePaths()));
 +
 +        sstable.selfRef().release();
 +    }
 +
 +    @Test
 +    public void testCommitOnlyOld() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier tidier = log.obsoleted(sstable);
 +        assertNotNull(tidier);
 +
 +        log.finish();
 +        sstable.markObsolete(tidier);
 +        sstable.selfRef().release();
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>());
 +    }
 +
 +    @Test
 +    public void testCommitMultipleFolders() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +
 +        File origiFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        File dataFolder1 = new File(origiFolder, "1");
 +        File dataFolder2 = new File(origiFolder, "2");
 +        Files.createDirectories(dataFolder1.toPath());
 +        Files.createDirectories(dataFolder2.toPath());
 +
 +        SSTableReader[] sstables = { sstable(dataFolder1, cfs, 0, 128),
 +                                     sstable(dataFolder1, cfs, 1, 128),
 +                                     sstable(dataFolder2, cfs, 2, 128),
 +                                     sstable(dataFolder2, cfs, 3, 128)
 +        };
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier[] tidiers = { log.obsoleted(sstables[0]), log.obsoleted(sstables[2]) };
 +
 +        log.trackNew(sstables[1]);
 +        log.trackNew(sstables[3]);
 +
 +        log.finish();
 +
 +        sstables[0].markObsolete(tidiers[0]);
 +        sstables[2].markObsolete(tidiers[1]);
 +
 +        Arrays.stream(sstables).forEach(s -> s.selfRef().release());
 +        LogTransaction.waitForDeletions();
 +
 +        assertFiles(dataFolder1.getPath(), new HashSet<>(sstables[1].getAllFilePaths()));
 +        assertFiles(dataFolder2.getPath(), new HashSet<>(sstables[3].getAllFilePaths()));
 +    }
 +
 +    @Test
 +    public void testAbortOnlyNew() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstable);
 +        log.abort();
 +
 +        sstable.selfRef().release();
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>());
 +    }
 +
 +    @Test
 +    public void testAbortOnlyOld() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier tidier = log.obsoleted(sstable);
 +        assertNotNull(tidier);
 +
 +        tidier.abort();
 +        log.abort();
 +
 +        sstable.selfRef().release();
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>(sstable.getAllFilePaths()));
 +    }
 +
 +    @Test
 +    public void testAbortMultipleFolders() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +
 +        File origiFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        File dataFolder1 = new File(origiFolder, "1");
 +        File dataFolder2 = new File(origiFolder, "2");
 +        Files.createDirectories(dataFolder1.toPath());
 +        Files.createDirectories(dataFolder2.toPath());
 +
 +        SSTableReader[] sstables = { sstable(dataFolder1, cfs, 0, 128),
 +                                     sstable(dataFolder1, cfs, 1, 128),
 +                                     sstable(dataFolder2, cfs, 2, 128),
 +                                     sstable(dataFolder2, cfs, 3, 128)
 +        };
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier[] tidiers = { log.obsoleted(sstables[0]), log.obsoleted(sstables[2]) };
 +
 +        log.trackNew(sstables[1]);
 +        log.trackNew(sstables[3]);
 +
 +        Arrays.stream(tidiers).forEach(LogTransaction.SSTableTidier::abort);
 +        log.abort();
 +
 +        Arrays.stream(sstables).forEach(s -> s.selfRef().release());
 +        LogTransaction.waitForDeletions();
 +
 +        assertFiles(dataFolder1.getPath(), new HashSet<>(sstables[0].getAllFilePaths()));
 +        assertFiles(dataFolder2.getPath(), new HashSet<>(sstables[2].getAllFilePaths()));
 +    }
 +
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_abort() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableOld = sstable(dataFolder, cfs, 0, 128);
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        // simulate tracking sstables with a failed transaction (new log file NOT deleted)
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +        LogTransaction.SSTableTidier tidier = log.obsoleted(sstableOld);
 +
 +        Set<File> tmpFiles = sstableNew.getAllFilePaths().stream().map(File::new).collect(Collectors.toSet());
 +
 +        sstableNew.selfRef().release();
 +        sstableOld.selfRef().release();
 +
-         Assert.assertEquals(tmpFiles, LogAwareFileLister.getTemporaryFiles(sstableNew.descriptor.directory));
++        Assert.assertEquals(tmpFiles, getTemporaryFiles(sstableNew.descriptor.directory));
 +
 +        // normally called at startup
 +        LogTransaction.removeUnfinishedLeftovers(cfs.metadata);
 +
 +        // sstableOld should be only table left
 +        Directories directories = new Directories(cfs.metadata);
 +        Map<Descriptor, Set<Component>> sstables = directories.sstableLister(Directories.OnTxnErr.THROW).list();
 +        assertEquals(1, sstables.size());
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>(sstableOld.getAllFilePaths()));
 +
 +        // complete the transaction before releasing files
 +        tidier.run();
 +        log.close();
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_commit() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableOld = sstable(dataFolder, cfs, 0, 128);
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        // simulate tracking sstables with a committed transaction (new log file deleted)
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +        LogTransaction.SSTableTidier tidier = log.obsoleted(sstableOld);
 +
 +        //Fake a commit
 +        log.txnFile().commit();
 +
 +        Set<File> tmpFiles = sstableOld.getAllFilePaths().stream().map(File::new).collect(Collectors.toSet());
 +
 +        sstableNew.selfRef().release();
 +        sstableOld.selfRef().release();
 +
-         Assert.assertEquals(tmpFiles, LogAwareFileLister.getTemporaryFiles(sstableOld.descriptor.directory));
++        Assert.assertEquals(tmpFiles, getTemporaryFiles(sstableOld.descriptor.directory));
 +
 +        // normally called at startup
 +        LogTransaction.removeUnfinishedLeftovers(cfs.metadata);
 +
 +        // sstableNew should be only table left
 +        Directories directories = new Directories(cfs.metadata);
 +        Map<Descriptor, Set<Component>> sstables = directories.sstableLister(Directories.OnTxnErr.THROW).list();
 +        assertEquals(1, sstables.size());
 +
 +        assertFiles(dataFolder.getPath(), new HashSet<>(sstableNew.getAllFilePaths()));
 +
 +        // complete the transaction to avoid LEAK errors
 +        tidier.run();
 +        assertNull(log.complete(null));
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_commit_multipleFolders() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +
 +        File origiFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        File dataFolder1 = new File(origiFolder, "1");
 +        File dataFolder2 = new File(origiFolder, "2");
 +        Files.createDirectories(dataFolder1.toPath());
 +        Files.createDirectories(dataFolder2.toPath());
 +
 +        SSTableReader[] sstables = { sstable(dataFolder1, cfs, 0, 128),
 +                                     sstable(dataFolder1, cfs, 1, 128),
 +                                     sstable(dataFolder2, cfs, 2, 128),
 +                                     sstable(dataFolder2, cfs, 3, 128)
 +        };
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier[] tidiers = { log.obsoleted(sstables[0]), log.obsoleted(sstables[2]) };
 +
 +        log.trackNew(sstables[1]);
 +        log.trackNew(sstables[3]);
 +
 +        Collection<File> logFiles = log.logFiles();
 +        Assert.assertEquals(2, logFiles.size());
 +
 +        // fake a commit
 +        log.txnFile().commit();
 +
 +        Arrays.stream(sstables).forEach(s -> s.selfRef().release());
 +
 +        // test listing
 +        Assert.assertEquals(sstables[0].getAllFilePaths().stream().map(File::new).collect(Collectors.toSet()),
-                             LogAwareFileLister.getTemporaryFiles(dataFolder1));
++                            getTemporaryFiles(dataFolder1));
 +        Assert.assertEquals(sstables[2].getAllFilePaths().stream().map(File::new).collect(Collectors.toSet()),
-                             LogAwareFileLister.getTemporaryFiles(dataFolder2));
++                            getTemporaryFiles(dataFolder2));
 +
 +        // normally called at startup
 +        LogTransaction.removeUnfinishedLeftovers(Arrays.asList(dataFolder1, dataFolder2));
 +
 +        // new tables should be only table left
 +        assertFiles(dataFolder1.getPath(), new HashSet<>(sstables[1].getAllFilePaths()));
 +        assertFiles(dataFolder2.getPath(), new HashSet<>(sstables[3].getAllFilePaths()));
 +
 +        // complete the transaction to avoid LEAK errors
 +        Arrays.stream(tidiers).forEach(LogTransaction.SSTableTidier::run);
 +        assertNull(log.complete(null));
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_abort_multipleFolders() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +
 +        File origiFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        File dataFolder1 = new File(origiFolder, "1");
 +        File dataFolder2 = new File(origiFolder, "2");
 +        Files.createDirectories(dataFolder1.toPath());
 +        Files.createDirectories(dataFolder2.toPath());
 +
 +        SSTableReader[] sstables = { sstable(dataFolder1, cfs, 0, 128),
 +                                     sstable(dataFolder1, cfs, 1, 128),
 +                                     sstable(dataFolder2, cfs, 2, 128),
 +                                     sstable(dataFolder2, cfs, 3, 128)
 +        };
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier[] tidiers = { log.obsoleted(sstables[0]), log.obsoleted(sstables[2]) };
 +
 +        log.trackNew(sstables[1]);
 +        log.trackNew(sstables[3]);
 +
 +        Collection<File> logFiles = log.logFiles();
 +        Assert.assertEquals(2, logFiles.size());
 +
 +        // fake an abort
 +        log.txnFile().abort();
 +
 +        Arrays.stream(sstables).forEach(s -> s.selfRef().release());
 +
 +        // test listing
 +        Assert.assertEquals(sstables[1].getAllFilePaths().stream().map(File::new).collect(Collectors.toSet()),
-                             LogAwareFileLister.getTemporaryFiles(dataFolder1));
++                            getTemporaryFiles(dataFolder1));
 +        Assert.assertEquals(sstables[3].getAllFilePaths().stream().map(File::new).collect(Collectors.toSet()),
-                             LogAwareFileLister.getTemporaryFiles(dataFolder2));
++                            getTemporaryFiles(dataFolder2));
 +
 +        // normally called at startup
 +        LogTransaction.removeUnfinishedLeftovers(Arrays.asList(dataFolder1, dataFolder2));
 +
 +        // old tables should be only table left
 +        assertFiles(dataFolder1.getPath(), new HashSet<>(sstables[0].getAllFilePaths()));
 +        assertFiles(dataFolder2.getPath(), new HashSet<>(sstables[2].getAllFilePaths()));
 +
 +        // complete the transaction to avoid LEAK errors
 +        Arrays.stream(tidiers).forEach(LogTransaction.SSTableTidier::run);
 +        assertNull(log.complete(null));
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_mismatchedFinalRecords() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert mismatched records
 +            FileUtils.append(logFiles.get(0), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +            FileUtils.append(logFiles.get(1), LogRecord.makeAbort(System.currentTimeMillis()).raw);
 +
 +        }, false);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_partialFinalRecords_first() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert a full record and a partial one
 +            String finalRecord = LogRecord.makeCommit(System.currentTimeMillis()).raw;
 +            int toChop = finalRecord.length() / 2;
 +            FileUtils.append(logFiles.get(0), finalRecord.substring(0, finalRecord.length() - toChop));
 +            FileUtils.append(logFiles.get(1), finalRecord);
 +
 +        }, true);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_partialFinalRecords_second() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert a full record and a partial one
 +            String finalRecord = LogRecord.makeCommit(System.currentTimeMillis()).raw;
 +            int toChop = finalRecord.length() / 2;
 +            FileUtils.append(logFiles.get(0), finalRecord);
 +            FileUtils.append(logFiles.get(1), finalRecord.substring(0, finalRecord.length() - toChop));
 +
 +        }, true);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_partialNonFinalRecord_first() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert a partial sstable record and a full commit record
 +            String sstableRecord = LogRecord.make(LogRecord.Type.ADD, Collections.emptyList(), 0, "abc").raw;
 +            int toChop = sstableRecord.length() / 2;
 +            FileUtils.append(logFiles.get(0), sstableRecord.substring(0, sstableRecord.length() - toChop));
 +            FileUtils.append(logFiles.get(1), sstableRecord);
 +            String finalRecord = LogRecord.makeCommit(System.currentTimeMillis()).raw;
 +            FileUtils.append(logFiles.get(0), finalRecord);
 +            FileUtils.append(logFiles.get(1), finalRecord);
 +
 +        }, false);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_partialNonFinalRecord_second() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert a partial sstable record and a full commit record
 +            String sstableRecord = LogRecord.make(LogRecord.Type.ADD, Collections.emptyList(), 0, "abc").raw;
 +            int toChop = sstableRecord.length() / 2;
 +            FileUtils.append(logFiles.get(0), sstableRecord);
 +            FileUtils.append(logFiles.get(1), sstableRecord.substring(0, sstableRecord.length() - toChop));
 +            String finalRecord = LogRecord.makeCommit(System.currentTimeMillis()).raw;
 +            FileUtils.append(logFiles.get(0), finalRecord);
 +            FileUtils.append(logFiles.get(1), finalRecord);
 +
 +        }, false);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_missingFinalRecords_first() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert only one commit record
 +            FileUtils.append(logFiles.get(0), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +
 +        }, true);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_missingFinalRecords_second() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert only one commit record
 +            FileUtils.append(logFiles.get(1), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +
 +        }, true);
 +    }
 +
 +    @Test
 +    public void testRemoveUnfinishedLeftovers_multipleFolders_tooManyFinalRecords() throws Throwable
 +    {
 +        testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(txn -> {
 +            List<File> logFiles = txn.logFiles();
 +            Assert.assertEquals(2, logFiles.size());
 +
 +            // insert mismatched records
 +            FileUtils.append(logFiles.get(0), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +            FileUtils.append(logFiles.get(1), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +            FileUtils.append(logFiles.get(1), LogRecord.makeCommit(System.currentTimeMillis()).raw);
 +
 +        }, false);
 +    }
 +
 +    private static void testRemoveUnfinishedLeftovers_multipleFolders_errorConditions(Consumer<LogTransaction> modifier, boolean shouldCommit) throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +
 +        File origiFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        File dataFolder1 = new File(origiFolder, "1");
 +        File dataFolder2 = new File(origiFolder, "2");
 +        Files.createDirectories(dataFolder1.toPath());
 +        Files.createDirectories(dataFolder2.toPath());
 +
 +        SSTableReader[] sstables = { sstable(dataFolder1, cfs, 0, 128),
 +                                     sstable(dataFolder1, cfs, 1, 128),
 +                                     sstable(dataFolder2, cfs, 2, 128),
 +                                     sstable(dataFolder2, cfs, 3, 128)
 +        };
 +
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        LogTransaction.SSTableTidier[] tidiers = { log.obsoleted(sstables[0]), log.obsoleted(sstables[2]) };
 +
 +        log.trackNew(sstables[1]);
 +        log.trackNew(sstables[3]);
 +
 +        // fake some error condition on the txn logs
 +        modifier.accept(log);
 +
 +        Arrays.stream(sstables).forEach(s -> s.selfRef().release());
 +
 +        LogTransaction.removeUnfinishedLeftovers(Arrays.asList(dataFolder1, dataFolder2));
 +        LogTransaction.waitForDeletions();
 +
 +        if (shouldCommit)
 +        {
 +            // only new sstables should still be there
 +            assertFiles(dataFolder1.getPath(), new HashSet<>(sstables[1].getAllFilePaths()));
 +            assertFiles(dataFolder2.getPath(), new HashSet<>(sstables[3].getAllFilePaths()));
 +        }
 +        else
 +        {
 +            // all files should still be there
 +            assertFiles(dataFolder1.getPath(), Sets.newHashSet(Iterables.concat(sstables[0].getAllFilePaths(),
 +                                                                                sstables[1].getAllFilePaths(),
 +                                                                                Collections.singleton(log.logFilePaths().get(0)))));
 +            assertFiles(dataFolder2.getPath(), Sets.newHashSet(Iterables.concat(sstables[2].getAllFilePaths(),
 +                                                                                sstables[3].getAllFilePaths(),
 +                                                                                Collections.singleton(log.logFilePaths().get(1)))));
 +        }
 +
 +
 +        // complete the transaction to avoid LEAK errors
 +        Arrays.stream(tidiers).forEach(LogTransaction.SSTableTidier::run);
 +        log.txnFile().commit(); // just anything to make sure transaction tidier will finish
 +        assertNull(log.complete(null));
 +    }
 +
 +    @Test
 +    public void testGetTemporaryFiles() throws IOException
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable1 = sstable(dataFolder, cfs, 0, 128);
 +
-         Set<File> tmpFiles = LogAwareFileLister.getTemporaryFiles(dataFolder);
++        Set<File> tmpFiles = getTemporaryFiles(dataFolder);
 +        assertNotNull(tmpFiles);
 +        assertEquals(0, tmpFiles.size());
 +
 +        try(LogTransaction log = new LogTransaction(OperationType.WRITE))
 +        {
 +            Directories directories = new Directories(cfs.metadata);
 +
 +            File[] beforeSecondSSTable = dataFolder.listFiles(pathname -> !pathname.isDirectory());
 +
 +            SSTableReader sstable2 = sstable(dataFolder, cfs, 1, 128);
 +            log.trackNew(sstable2);
 +
 +            Map<Descriptor, Set<Component>> sstables = directories.sstableLister(Directories.OnTxnErr.THROW).list();
 +            assertEquals(2, sstables.size());
 +
 +            // this should contain sstable1, sstable2 and the transaction log file
 +            File[] afterSecondSSTable = dataFolder.listFiles(pathname -> !pathname.isDirectory());
 +
 +            int numNewFiles = afterSecondSSTable.length - beforeSecondSSTable.length;
 +            assertEquals(numNewFiles - 1, sstable2.getAllFilePaths().size()); // new files except for transaction log file
 +
-             tmpFiles = LogAwareFileLister.getTemporaryFiles(dataFolder);
++            tmpFiles = getTemporaryFiles(dataFolder);
 +            assertNotNull(tmpFiles);
 +            assertEquals(numNewFiles - 1, tmpFiles.size());
 +
 +            File ssTable2DataFile = new File(sstable2.descriptor.filenameFor(Component.DATA));
 +            File ssTable2IndexFile = new File(sstable2.descriptor.filenameFor(Component.PRIMARY_INDEX));
 +
 +            assertTrue(tmpFiles.contains(ssTable2DataFile));
 +            assertTrue(tmpFiles.contains(ssTable2IndexFile));
 +
 +            List<File> files = directories.sstableLister(Directories.OnTxnErr.THROW).listFiles();
 +            List<File> filesNoTmp = directories.sstableLister(Directories.OnTxnErr.THROW).skipTemporary(true).listFiles();
 +            assertNotNull(files);
 +            assertNotNull(filesNoTmp);
 +
 +            assertTrue(files.contains(ssTable2DataFile));
 +            assertTrue(files.contains(ssTable2IndexFile));
 +
 +            assertFalse(filesNoTmp.contains(ssTable2DataFile));
 +            assertFalse(filesNoTmp.contains(ssTable2IndexFile));
 +
 +            log.finish();
 +
 +            //Now it should be empty since the transaction has finished
-             tmpFiles = LogAwareFileLister.getTemporaryFiles(dataFolder);
++            tmpFiles = getTemporaryFiles(dataFolder);
 +            assertNotNull(tmpFiles);
 +            assertEquals(0, tmpFiles.size());
 +
 +            filesNoTmp = directories.sstableLister(Directories.OnTxnErr.THROW).skipTemporary(true).listFiles();
 +            assertNotNull(filesNoTmp);
 +            assertTrue(filesNoTmp.contains(ssTable2DataFile));
 +            assertTrue(filesNoTmp.contains(ssTable2IndexFile));
 +
 +            sstable1.selfRef().release();
 +            sstable2.selfRef().release();
 +        }
 +    }
 +
 +    @Test
 +    public void testWrongChecksumLastLine() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          { // Fake a commit with invalid checksum
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d,0,0][%d]", now, 12345678L)));
 +                          },
 +                          true);
 +    }
 +
 +    @Test
 +    public void testWrongChecksumSecondFromLastLine() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          { // Fake two lines with invalid checksum
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("add:[ma-3-big,%d,4][%d]", now, 12345678L)));
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d,0,0][%d]", now, 12345678L)));
 +                          },
 +                          false);
 +    }
 +
 +    @Test
 +    public void testWrongChecksumLastLineMissingFile() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          { // Fake a commit with invalid checksum and also delete one of the old files
 +                              for (String filePath : s.getAllFilePaths())
 +                              {
 +                                  if (filePath.endsWith("Data.db"))
 +                                  {
 +                                      assertTrue(FileUtils.delete(filePath));
 +                                      assertNull(t.txnFile().syncFolder(null));
 +                                      break;
 +                                  }
 +                              }
 +
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d,0,0][%d]", now, 12345678L)));
 +                          },
 +                          false);
 +    }
 +
 +    @Test
 +    public void testWrongChecksumLastLineWrongRecordFormat() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          { // Fake a commit with invalid checksum and a wrong record format (extra spaces)
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d ,0 ,0 ][%d]", now, 12345678L)));
 +                          },
 +                          true);
 +    }
 +
 +    @Test
 +    public void testMissingChecksumLastLine() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          {
 +                              // Fake a commit without a checksum
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d,0,0]", now)));
 +                          },
 +                          true);
 +    }
 +
 +    @Test
 +    public void testMissingChecksumSecondFromLastLine() throws IOException
 +    {
 +        testCorruptRecord((t, s) ->
 +                          { // Fake two lines without a checksum
 +                              long now = System.currentTimeMillis();
 +                              t.logFiles().forEach( f -> FileUtils.append(f, String.format("add:[ma-3-big,%d,4]", now)));
 +                              t.logFiles().forEach(f -> FileUtils.append(f, String.format("commit:[%d,0,0]", now)));
 +                          },
 +                          false);
 +    }
 +
 +    @Test
 +    public void testUnparsableLastRecord() throws IOException
 +    {
 +        testCorruptRecord((t, s) -> t.logFiles().forEach(f -> FileUtils.append(f, "commit:[a,b,c][12345678]")), true);
 +    }
 +
 +    @Test
 +    public void testUnparsableFirstRecord() throws IOException
 +    {
 +        testCorruptRecord((t, s) -> t.logFiles().forEach(f -> {
 +            List<String> lines = FileUtils.readLines(f);
 +            lines.add(0, "add:[a,b,c][12345678]");
 +            FileUtils.replace(f, lines.toArray(new String[lines.size()]));
 +        }), false);
 +    }
 +
 +    private static void testCorruptRecord(BiConsumer<LogTransaction, SSTableReader> modifier, boolean isRecoverable) throws IOException
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableOld = sstable(dataFolder, cfs, 0, 128);
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        // simulate tracking sstables with a committed transaction except the checksum will be wrong
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +        log.obsoleted(sstableOld);
 +
 +        // Modify the transaction log or disk state for sstableOld
 +        modifier.accept(log, sstableOld);
 +
 +        assertNull(log.complete(null));
 +
 +        sstableOld.selfRef().release();
 +        sstableNew.selfRef().release();
 +
 +        // The files on disk, for old files make sure to exclude the files that were deleted by the modifier
 +        Set<String> newFiles = sstableNew.getAllFilePaths().stream().collect(Collectors.toSet());
 +        Set<String> oldFiles = sstableOld.getAllFilePaths().stream().filter(p -> new File(p).exists()).collect(Collectors.toSet());
 +
 +        //This should filter as in progress since the last record is corrupt
-         assertFiles(newFiles, LogAwareFileLister.getTemporaryFiles(dataFolder));
-         assertFiles(oldFiles, LogAwareFileLister.getFinalFiles(dataFolder));
++        assertFiles(newFiles, getTemporaryFiles(dataFolder));
++        assertFiles(oldFiles, getFinalFiles(dataFolder));
 +
 +        if (isRecoverable)
 +        { // the corruption is recoverable but the commit record is unreadable so the transaction is still in progress
 +
 +            //This should remove new files
 +            LogTransaction.removeUnfinishedLeftovers(cfs.metadata);
 +
 +            // make sure to exclude the old files that were deleted by the modifier
 +            assertFiles(dataFolder.getPath(), oldFiles);
 +        }
 +        else
 +        { // if an intermediate line was also modified, it should ignore the tx log file
 +
 +            //This should not remove any files
 +            LogTransaction.removeUnfinishedLeftovers(cfs.metadata);
 +
 +            assertFiles(dataFolder.getPath(), Sets.newHashSet(Iterables.concat(newFiles,
 +                                                                               oldFiles,
 +                                                                               log.logFilePaths())));
 +        }
 +    }
 +
 +    @Test
 +    public void testObsoletedDataFileUpdateTimeChanged() throws IOException
 +    {
 +        testObsoletedFilesChanged(sstable ->
 +                                  {
 +                                      // increase the modification time of the Data file
 +                                      for (String filePath : sstable.getAllFilePaths())
 +                                      {
 +                                          if (filePath.endsWith("Data.db"))
 +                                              assertTrue(new File(filePath).setLastModified(System.currentTimeMillis() + 60000)); //one minute later
 +                                      }
 +                                  });
 +    }
 +
 +    private static void testObsoletedFilesChanged(Consumer<SSTableReader> modifier) throws IOException
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstableOld = sstable(dataFolder, cfs, 0, 128);
 +        SSTableReader sstableNew = sstable(dataFolder, cfs, 1, 128);
 +
 +        // simulate tracking sstables with a committed transaction except the checksum will be wrong
 +        LogTransaction log = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(log);
 +
 +        log.trackNew(sstableNew);
 +        /*TransactionLog.SSTableTidier tidier =*/ log.obsoleted(sstableOld);
 +
 +        //modify the old sstable files
 +        modifier.accept(sstableOld);
 +
 +        //Fake a commit
 +        log.txnFile().commit();
 +
 +        //This should not remove the old files
 +        LogTransaction.removeUnfinishedLeftovers(cfs.metadata);
 +
 +        assertFiles(dataFolder.getPath(), Sets.newHashSet(Iterables.concat(
 +                                                                          sstableNew.getAllFilePaths(),
 +                                                                          sstableOld.getAllFilePaths(),
 +                                                                          log.logFilePaths())));
 +
 +        sstableOld.selfRef().release();
 +        sstableNew.selfRef().release();
 +
 +        // complete the transaction to avoid LEAK errors
 +        assertNull(log.complete(null));
 +
 +        assertFiles(dataFolder.getPath(), Sets.newHashSet(Iterables.concat(sstableNew.getAllFilePaths(),
 +                                                                           sstableOld.getAllFilePaths(),
 +                                                                           log.logFilePaths())));
 +    }
 +
 +    @Test
 +    public void testGetTemporaryFilesSafeAfterObsoletion() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction logs = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(logs);
 +
 +        LogTransaction.SSTableTidier tidier = logs.obsoleted(sstable);
 +
 +        logs.finish();
 +
 +        sstable.markObsolete(tidier);
 +        sstable.selfRef().release();
 +
 +        // This should race with the asynchronous deletion of txn log files
 +        // It doesn't matter what it returns but it should not throw because the txn
 +        // was completed before deleting files (i.e. releasing sstables)
 +        for (int i = 0; i < 200; i++)
-             LogAwareFileLister.getTemporaryFiles(dataFolder);
++            getTemporaryFiles(dataFolder);
 +    }
 +
 +    @Test
 +    public void testGetTemporaryFilesThrowsIfCompletingAfterObsoletion() throws Throwable
 +    {
 +        ColumnFamilyStore cfs = MockSchema.newCFS(KEYSPACE);
 +        File dataFolder = new Directories(cfs.metadata).getDirectoryForNewSSTables();
 +        SSTableReader sstable = sstable(dataFolder, cfs, 0, 128);
 +
 +        LogTransaction logs = new LogTransaction(OperationType.COMPACTION);
 +        assertNotNull(logs);
 +
 +        LogTransaction.SSTableTidier tidier = logs.obsoleted(sstable);
 +
 +        sstable.markObsolete(tidier);
 +        sstable.selfRef().release();
 +
 +        LogTransaction.waitForDeletions();
 +
 +        try
 +        {
 +            // This should race with the asynchronous deletion of txn log files
 +            // it should throw because we are violating the requirement that a transaction must
 +            // finish before deleting files (i.e. releasing sstables)
-             LogAwareFileLister.getTemporaryFiles(dataFolder);
++            getTemporaryFiles(dataFolder);
 +            fail("Expected runtime exception");
 +        }
 +        catch(RuntimeException e)
 +        {
 +            //pass as long as the cause is not an assertion
 +            assertFalse(e.getCause() instanceof AssertionError);
 +        }
 +
 +        logs.finish();
 +    }
 +
 +    private static SSTableReader sstable(File dataFolder, ColumnFamilyStore cfs, int generation, int size) throws IOException
 +    {
 +        Descriptor descriptor = new Descriptor(dataFolder, cfs.keyspace.getName(), cfs.getTableName(), generation);
 +        Set<Component> components = ImmutableSet.of(Component.DATA, Component.PRIMARY_INDEX, Component.FILTER, Component.TOC);
 +        for (Component component : components)
 +        {
 +            File file = new File(descriptor.filenameFor(component));
 +            if (!file.exists())
 +                assertTrue(file.createNewFile());
 +            try (RandomAccessFile raf = new RandomAccessFile(file, "rw"))
 +            {
 +                raf.setLength(size);
 +            }
 +        }
 +
 +        SegmentedFile dFile = new BufferedSegmentedFile(new ChannelProxy(new File(descriptor.filenameFor(Component.DATA))), RandomAccessReader.DEFAULT_BUFFER_SIZE, 0);
 +        SegmentedFile iFile = new BufferedSegmentedFile(new ChannelProxy(new File(descriptor.filenameFor(Component.PRIMARY_INDEX))), RandomAccessReader.DEFAULT_BUFFER_SIZE, 0);
 +
 +        SerializationHeader header = SerializationHeader.make(cfs.metadata, Collections.emptyList());
 +        StatsMetadata metadata = (StatsMetadata) new MetadataCollector(cfs.metadata.comparator)
 +                                                 .finalizeMetadata(cfs.metadata.partitioner.getClass().getCanonicalName(), 0.01f, -1, header)
 +                                                 .get(MetadataType.STATS);
 +        SSTableReader reader = SSTableReader.internalOpen(descriptor,
 +                                                          components,
 +                                                          cfs.metadata,
 +                                                          dFile,
 +                                                          iFile,
 +                                                          MockSchema.indexSummary.sharedCopy(),
 +                                                          new AlwaysPresentFilter(),
 +                                                          1L,
 +                                                          metadata,
 +                                                          SSTableReader.OpenReason.NORMAL,
 +                                                          header);
 +        reader.first = reader.last = MockSchema.readerBounds(generation);
 +        return reader;
 +    }
 +
-     private static void assertFiles(String dirPath, Set<String> expectedFiles)
++    private static void assertFiles(String dirPath, Set<String> expectedFiles) throws IOException
 +    {
 +        assertFiles(dirPath, expectedFiles, false);
 +    }
 +
-     private static void assertFiles(String dirPath, Set<String> expectedFiles, boolean excludeNonExistingFiles)
++    private static void assertFiles(String dirPath, Set<String> expectedFiles, boolean excludeNonExistingFiles) throws IOException
 +    {
 +        LogTransaction.waitForDeletions();
 +
-         File dir = new File(dirPath);
++        File dir = new File(dirPath).getCanonicalFile();
 +        File[] files = dir.listFiles();
 +        if (files != null)
 +        {
 +            for (File file : files)
 +            {
 +                if (file.isDirectory())
 +                    continue;
 +
 +                String filePath = file.getPath();
 +                assertTrue(String.format("%s not in [%s]", filePath, expectedFiles), expectedFiles.contains(filePath));
 +                expectedFiles.remove(filePath);
 +            }
 +        }
 +
 +        if (excludeNonExistingFiles)
 +        {
 +            for (String filePath : expectedFiles)
 +            {
 +                File file = new File(filePath);
 +                if (!file.exists())
 +                    expectedFiles.remove(filePath);
 +            }
 +        }
 +
 +        assertTrue(expectedFiles.toString(), expectedFiles.isEmpty());
 +    }
 +
 +    // Check either that a temporary file is expected to exist (in the existingFiles) or that
 +    // it does not exist any longer (on Windows we need to check File.exists() because a list
 +    // might return a file as existing even if it does not)
 +    private static void assertFiles(Iterable<String> existingFiles, Set<File> temporaryFiles)
 +    {
 +        for (String filePath : existingFiles)
 +        {
 +            File file = new File(filePath);
 +            assertTrue(filePath, temporaryFiles.contains(file));
 +            temporaryFiles.remove(file);
 +        }
 +
 +        for (File file : temporaryFiles)
 +        {
 +            if (!file.exists())
 +                temporaryFiles.remove(file);
 +        }
 +
 +        assertTrue(temporaryFiles.toString(), temporaryFiles.isEmpty());
 +    }
++
++    static Set<File> getTemporaryFiles(File folder)
++    {
++        return listFiles(folder, Directories.FileType.TEMPORARY);
++    }
++
++    static Set<File> getFinalFiles(File folder)
++    {
++        return listFiles(folder, Directories.FileType.FINAL);
++    }
++
++    static Set<File> listFiles(File folder, Directories.FileType... types)
++    {
++        Collection<Directories.FileType> match = Arrays.asList(types);
++        return new LogAwareFileLister(folder.toPath(),
++                                      (file, type) -> match.contains(type),
++                                      Directories.OnTxnErr.IGNORE).list()
++                       .stream()
++                       .map(f -> {
++                           try
++                           {
++                               return f.getCanonicalFile();
++                           }
++                           catch (IOException e)
++                           {
++                               throw new IOError(e);
++                           }
++                       })
++                       .collect(Collectors.toSet());
++    }
 +}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/15131391/test/unit/org/apache/cassandra/io/sstable/DescriptorTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/io/sstable/DescriptorTest.java
index 19eca40,6354fc2..184d637
--- a/test/unit/org/apache/cassandra/io/sstable/DescriptorTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/DescriptorTest.java
@@@ -116,18 -127,33 +115,33 @@@ public class DescriptorTes
      }
  
      @Test
-     public void validateNames()
+     public void testEquality()
      {
+         // Descriptor should be equal when parent directory points to the same directory
+         File dir = new File(".");
 -        Descriptor desc1 = new Descriptor(dir, "ks", "cf", 1, Descriptor.Type.FINAL);
 -        Descriptor desc2 = new Descriptor(dir.getAbsoluteFile(), "ks", "cf", 1, Descriptor.Type.FINAL);
++        Descriptor desc1 = new Descriptor(dir, "ks", "cf", 1);
++        Descriptor desc2 = new Descriptor(dir.getAbsoluteFile(), "ks", "cf", 1);
+         assertEquals(desc1, desc2);
+         assertEquals(desc1.hashCode(), desc2.hashCode());
+     }
  
-         String names[] = {
-             /*"system-schema_keyspaces-ka-1-CompressionInfo.db",  "system-schema_keyspaces-ka-1-Summary.db",
-             "system-schema_keyspaces-ka-1-Data.db",             "system-schema_keyspaces-ka-1-TOC.txt",
-             "system-schema_keyspaces-ka-1-Digest.sha1",         "system-schema_keyspaces-ka-2-CompressionInfo.db",
-             "system-schema_keyspaces-ka-1-Filter.db",           "system-schema_keyspaces-ka-2-Data.db",
-             "system-schema_keyspaces-ka-1-Index.db",            "system-schema_keyspaces-ka-2-Digest.sha1",
-             "system-schema_keyspaces-ka-1-Statistics.db",
-             "system-schema_keyspacest-tmp-ka-1-Data.db",*/
-             "system-schema_keyspace-ka-1-"+ SSTableFormat.Type.BIG.name+"-Data.db"
+     @Test
+     public void validateNames()
+     {
 -
++        // TODO tmp file name probably is not handled correctly after CASSANDRA-7066
+         String[] names = {
+              // old formats
+              "system-schema_keyspaces-jb-1-Data.db",
 -             "system-schema_keyspaces-tmp-jb-1-Data.db",
++             //"system-schema_keyspaces-tmp-jb-1-Data.db",
+              "system-schema_keyspaces-ka-1-big-Data.db",
 -             "system-schema_keyspaces-tmp-ka-1-big-Data.db",
++             //"system-schema_keyspaces-tmp-ka-1-big-Data.db",
+              // 2ndary index
+              "keyspace1-standard1.idx1-ka-1-big-Data.db",
+              // new formats
+              "la-1-big-Data.db",
 -             "tmp-la-1-big-Data.db",
++             //"tmp-la-1-big-Data.db",
+              // 2ndary index
+              ".idx1" + File.separator + "la-1-big-Data.db",
          };
  
          for (String name : names)


Mime
View raw message