Merge 94c579211b into bdbdf8d710

src/java/org/apache/cassandra/journal/Compactor.java

@@ -20,6 +20,7 @@ package org.apache.cassandra.journal;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -29,6 +30,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
 import org.apache.cassandra.concurrent.Shutdownable;
+import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.concurrent.WaitQueue;
 
 import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
@@ -87,6 +89,24 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
             toCompact.subList(limit, toCompact.size()).clear();
         }
 
+        // Hold refs on all segments to prevent them from being cleaned up (mmap unmapped)
+        // while the compactor is still reading from them. Segments whose ref can't be
+        // acquired have already been released and must be skipped.
+        List<Ref<Segment<K, V>>> refs = new ArrayList<>(toCompact.size());
+        Iterator<StaticSegment<K, V>> it = toCompact.iterator();
+        while (it.hasNext())
+        {
+            StaticSegment<K, V> segment = it.next();
+            Ref<Segment<K, V>> ref = segment.tryRef();
+            if (ref == null)
+                it.remove();
+            else
+                refs.add(ref);
+        }
+
+        if (toCompact.isEmpty())
+            return;
+
         try
         {
             Collection<StaticSegment<K, V>> newSegments = segmentCompactor.compact(toCompact);
@@ -104,6 +124,11 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
         {
             throw new RuntimeException("Could not compact segments: " + toCompact);
         }
+        finally
+        {
+            for (Ref<Segment<K, V>> ref : refs)
+                ref.release();
+        }
     }
 
     @Override