diff --git a/TimeWindowCompactionStrategy/src/main/java/com/jeffjirsa/cassandra/db/compaction/TimeWindowCompactionStrategy.java b/TimeWindowCompactionStrategy/src/main/java/com/jeffjirsa/cassandra/db/compaction/TimeWindowCompactionStrategy.java
index 5bf24bc..fb0a9b3 100644
--- a/TimeWindowCompactionStrategy/src/main/java/com/jeffjirsa/cassandra/db/compaction/TimeWindowCompactionStrategy.java
+++ b/TimeWindowCompactionStrategy/src/main/java/com/jeffjirsa/cassandra/db/compaction/TimeWindowCompactionStrategy.java
@@ -22,6 +22,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.Iterator;
 import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
@@ -34,7 +35,11 @@
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.sstable.Component;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -112,6 +117,7 @@ private List<SSTableReader> getNextBackgroundSSTables(final int gcBefore)
         Set<SSTableReader> candidates = Sets.newHashSet(filterSuspectSSTables(uncompacting));
 
         List<SSTableReader> compactionCandidates = new ArrayList<SSTableReader>(getNextNonExpiredSSTables(Sets.difference(candidates, expired), gcBefore));
+
         if (!expired.isEmpty())
         {
             logger.debug("Including expired sstables: {}", expired);
@@ -123,25 +129,34 @@ private List<SSTableReader> getNextBackgroundSSTables(final int gcBefore)
 
     private List<SSTableReader> getNextNonExpiredSSTables(Iterable<SSTableReader> nonExpiringSSTables, final int gcBefore)
     {
+        // First get the a list of tables that could drop tombstones.
+        List<SSTableReader> sstablesWithTombstones = new ArrayList<>();
+        for (SSTableReader sstable : nonExpiringSSTables)
+        {
+            if (worthDroppingTombstones(sstable, gcBefore))
+                sstablesWithTombstones.add(sstable);
+        }
+
+        // If there are files that could be compacted because they reached their tombstone ratio, then clean them up.
+        // NOTE: You should set a the tombstone ratio and tombstone compaction interval high enough so you're not constantly compacting SSTables.
+        if(!sstablesWithTombstones.isEmpty())
+        {
+            // Compact the tables with the highest tombstone ratios first.
+            return Collections.singletonList(Collections.max(sstablesWithTombstones, (o1, o2) ->
+                    Double.compare(o1.getEstimatedDroppableTombstoneRatio(gcBefore), o2.getEstimatedDroppableTombstoneRatio(gcBefore))));
+        }
+
+        // Now check for the most interesting tables.
         List<SSTableReader> mostInteresting = getCompactionCandidates(nonExpiringSSTables);
 
         if (mostInteresting != null)
         {
             return mostInteresting;
         }
-
-        // if there is no sstable to compact in standard way, try compacting single sstable whose droppable tombstone
-        // ratio is greater than threshold.
-        List<SSTableReader> sstablesWithTombstones = new ArrayList<>();
-        for (SSTableReader sstable : nonExpiringSSTables)
+        else
         {
-            if (worthDroppingTombstones(sstable, gcBefore))
-                sstablesWithTombstones.add(sstable);
-        }
-        if (sstablesWithTombstones.isEmpty())
             return Collections.emptyList();
-
-        return Collections.singletonList(Collections.min(sstablesWithTombstones, new SSTableReader.SizeComparator()));
+        }
     }
 
     private List<SSTableReader> getCompactionCandidates(Iterable<SSTableReader> candidateSSTables)
@@ -389,4 +404,4 @@ public String toString()
                 cfs.getMinimumCompactionThreshold(),
                 cfs.getMaximumCompactionThreshold());
     }
-}
+}
\ No newline at end of file