diff --git a/CHANGES.txt b/CHANGES.txt
index 7bec00d66f..6aeb6eac34 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 5.1
+ * Adopt Unsafe::invokeCleaner for Direct ByteBuffer cleaning (CASSANDRA-20677)
  * Add additional metrics around hints (CASSANDRA-20499)
  * Support for add and replace in IntervalTree (CASSANDRA-20513)
  * Enable single_sstable_uplevel by default for LCS (CASSANDRA-18509)
diff --git a/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java b/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java
index fb671130a1..7cc7eea7f9 100644
--- a/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java
+++ b/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java
@@ -27,10 +27,10 @@ import java.nio.file.StandardOpenOption;
 import net.openhft.chronicle.core.util.ThrowingFunction;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.FSWriteError;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.SimpleCachedBufferPool;
 import org.apache.cassandra.utils.NativeLibrary;
 import org.apache.cassandra.utils.SyncUtil;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 /*
  * Memory-mapped segment. Maps the destination channel into an appropriately-sized memory-mapped buffer in which the
@@ -108,7 +108,7 @@ public class MemoryMappedSegment extends CommitLogSegment
     @Override
     protected void internalClose()
     {
-        FileUtils.clean(buffer);
+        MemoryUtil.clean(buffer);
         super.internalClose();
     }
 
diff --git a/src/java/org/apache/cassandra/db/streaming/CompressedInputStream.java b/src/java/org/apache/cassandra/db/streaming/CompressedInputStream.java
index d0b1e4c3c1..d754756f77 100644
--- a/src/java/org/apache/cassandra/db/streaming/CompressedInputStream.java
+++ b/src/java/org/apache/cassandra/db/streaming/CompressedInputStream.java
@@ -30,10 +30,10 @@ import com.google.common.primitives.Ints;
 import org.apache.cassandra.io.compress.CompressionMetadata;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.DataInputPlus;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RebufferingInputStream;
 import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.utils.ChecksumType;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static java.lang.Math.max;
 import static java.lang.String.format;
@@ -146,7 +146,7 @@ public class CompressedInputStream extends RebufferingInputStream implements Aut
                 // with poorly compressible data, it's possible for a compressed chunk to be larger than
                 // configured uncompressed chunk size - depending on data, min_compress_ratio, and compressor;
                 // we may need to resize the compressed buffer.
-                FileUtils.clean(compressedChunk);
+                MemoryUtil.clean(compressedChunk);
                 compressedChunk = ByteBuffer.allocateDirect(max((int) (compressedChunk.capacity() * GROWTH_FACTOR), chunkLength));
             }
 
@@ -212,13 +212,13 @@ public class CompressedInputStream extends RebufferingInputStream implements Aut
     {
         if (null != buffer)
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             buffer = null;
         }
 
         if (null != compressedChunk)
         {
-            FileUtils.clean(compressedChunk);
+            MemoryUtil.clean(compressedChunk);
             compressedChunk = null;
         }
     }
diff --git a/src/java/org/apache/cassandra/db/tries/InMemoryTrie.java b/src/java/org/apache/cassandra/db/tries/InMemoryTrie.java
index 7ba7a0ae1c..a322d94a28 100644
--- a/src/java/org/apache/cassandra/db/tries/InMemoryTrie.java
+++ b/src/java/org/apache/cassandra/db/tries/InMemoryTrie.java
@@ -28,11 +28,11 @@ import com.google.common.annotations.VisibleForTesting;
 import org.agrona.concurrent.UnsafeBuffer;
 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.io.compress.BufferType;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.bytecomparable.ByteSource;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 import org.apache.cassandra.utils.ObjectSizes;
 
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.github.jamm.MemoryMeterStrategy;
 
 /**
@@ -191,7 +191,7 @@ public class InMemoryTrie<T> extends InMemoryReadTrie<T>
         for (UnsafeBuffer b : buffers)
         {
             if (b != null)
-                FileUtils.clean(b.byteBuffer());
+                MemoryUtil.clean(b.byteBuffer());
         }
     }
 
diff --git a/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java b/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java
index b57f898a74..c7bf42c755 100644
--- a/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java
+++ b/src/java/org/apache/cassandra/hints/ChecksummedDataInput.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.io.util.*;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.NativeLibrary;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 /**
  * A {@link RandomAccessReader} wrapper that calculates the CRC in place.
@@ -241,7 +242,7 @@ public class ChecksummedDataInput extends RebufferingInputStream
     @Override
     public void close()
     {
-        FileUtils.clean(buffer);
+        MemoryUtil.clean(buffer);
         channel.close();
     }
 
diff --git a/src/java/org/apache/cassandra/hints/HintsBuffer.java b/src/java/org/apache/cassandra/hints/HintsBuffer.java
index 41c9cdf234..6517b52384 100644
--- a/src/java/org/apache/cassandra/hints/HintsBuffer.java
+++ b/src/java/org/apache/cassandra/hints/HintsBuffer.java
@@ -32,10 +32,10 @@ import java.util.zip.CRC32;
 
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.DataOutputBufferFixed;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.utils.AbstractIterator;
 import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.utils.FBUtilities.updateChecksum;
 import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@@ -101,7 +101,7 @@ final class HintsBuffer
 
     void free()
     {
-        FileUtils.clean(slab);
+        MemoryUtil.clean(slab);
     }
 
     /**
diff --git a/src/java/org/apache/cassandra/index/sasi/utils/MappedBuffer.java b/src/java/org/apache/cassandra/index/sasi/utils/MappedBuffer.java
index 0899be6b2b..684ec26271 100644
--- a/src/java/org/apache/cassandra/index/sasi/utils/MappedBuffer.java
+++ b/src/java/org/apache/cassandra/index/sasi/utils/MappedBuffer.java
@@ -24,8 +24,8 @@ import java.nio.channels.FileChannel.MapMode;
 
 import org.apache.cassandra.db.marshal.AbstractType;
 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.utils.memory.MemoryUtil;
 
 import com.google.common.annotations.VisibleForTesting;
 
@@ -225,7 +225,7 @@ public class MappedBuffer implements Closeable
         try
         {
             for (MappedByteBuffer segment : pages)
-                FileUtils.clean(segment);
+                MemoryUtil.clean(segment);
         }
         catch (Exception e)
         {
diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
index 4a7f8161cc..ea6448f818 100644
--- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
+++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
@@ -32,11 +32,11 @@ import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 import org.apache.cassandra.io.util.ChecksumWriter;
 import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.io.util.File;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.io.util.SequentialWriterOption;
 import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.utils.Throwables.merge;
 
@@ -246,7 +246,7 @@ public class CompressedSequentialWriter extends SequentialWriter
         int chunkSize = (int) (metadataWriter.chunkOffsetBy(realMark.nextChunkIndex) - chunkOffset - 4);
         if (compressed.capacity() < chunkSize)
         {
-            FileUtils.clean(compressed);
+            MemoryUtil.clean(compressed);
             compressed = compressor.preferredBufferType().allocate(chunkSize);
         }
 
@@ -401,7 +401,10 @@ public class CompressedSequentialWriter extends SequentialWriter
             accumulate = super.doPreCleanup(accumulate);
             if (compressed != null)
             {
-                try { FileUtils.clean(compressed); }
+                try
+                {
+                    MemoryUtil.clean(compressed);
+                }
                 catch (Throwable t) { accumulate = merge(accumulate, t); }
                 compressed = null;
             }
diff --git a/src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java b/src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java
index 201eb619e8..62b7667ba3 100644
--- a/src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java
+++ b/src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java
@@ -323,7 +323,7 @@ public class BufferedDataOutputStreamPlus extends DataOutputStreamPlus
 
         doFlush(0);
         channel.close();
-        FileUtils.clean(buffer);
+        MemoryUtil.clean(buffer);
         buffer = null;
     }
 
diff --git a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java
index 837e10c1b6..308219aa5f 100644
--- a/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java
+++ b/src/java/org/apache/cassandra/io/util/DataOutputBuffer.java
@@ -29,6 +29,7 @@ import com.google.common.base.Preconditions;
 
 import io.netty.util.concurrent.FastThreadLocal;
 import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.DATA_OUTPUT_BUFFER_ALLOCATE_TYPE;
 import static org.apache.cassandra.config.CassandraRelevantProperties.DOB_DOUBLING_THRESHOLD_MB;
@@ -183,7 +184,8 @@ public class DataOutputBuffer extends BufferedDataOutputStreamPlus
 
     protected void setBuffer(ByteBuffer newBuffer)
     {
-        FileUtils.clean(buffer); // free if direct
+        // free if direct
+        MemoryUtil.clean(buffer);
         buffer = newBuffer;
     }
 
diff --git a/src/java/org/apache/cassandra/io/util/FileInputStreamPlus.java b/src/java/org/apache/cassandra/io/util/FileInputStreamPlus.java
index 2bd57a99d2..89e0cb378c 100644
--- a/src/java/org/apache/cassandra/io/util/FileInputStreamPlus.java
+++ b/src/java/org/apache/cassandra/io/util/FileInputStreamPlus.java
@@ -24,6 +24,8 @@ import java.nio.channels.FileChannel;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 
+import org.apache.cassandra.utils.memory.MemoryUtil;
+
 public class FileInputStreamPlus extends RebufferingInputStream
 {
     final FileChannel channel;
@@ -76,7 +78,7 @@ public class FileInputStreamPlus extends RebufferingInputStream
         {
             try
             {
-                FileUtils.clean(buffer);
+                MemoryUtil.clean(buffer);
             }
             finally
             {
diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java
index c8c605d7da..a67f70dd96 100644
--- a/src/java/org/apache/cassandra/io/util/FileUtils.java
+++ b/src/java/org/apache/cassandra/io/util/FileUtils.java
@@ -21,10 +21,6 @@ import java.io.BufferedWriter;
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
-import java.lang.invoke.MethodHandle;
-import java.lang.invoke.MethodHandles;
-import java.lang.reflect.Method;
-import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
@@ -75,39 +71,6 @@ public final class FileUtils
 
     private static final DecimalFormat df = new DecimalFormat("#.##");
 
-    private static final Class clsDirectBuffer;
-    private static final MethodHandle mhDirectBufferCleaner;
-    private static final MethodHandle mhCleanerClean;
-
-    static
-    {
-        try
-        {
-            clsDirectBuffer = Class.forName("sun.nio.ch.DirectBuffer");
-            Method mDirectBufferCleaner = clsDirectBuffer.getMethod("cleaner");
-            mhDirectBufferCleaner = MethodHandles.lookup().unreflect(mDirectBufferCleaner);
-            Method mCleanerClean = mDirectBufferCleaner.getReturnType().getMethod("clean");
-            mhCleanerClean = MethodHandles.lookup().unreflect(mCleanerClean);
-
-            ByteBuffer buf = ByteBuffer.allocateDirect(1);
-            clean(buf);
-        }
-        catch (IllegalAccessException e)
-        {
-            logger.error("FATAL: Cassandra is unable to access required classes. This usually means it has been " +
-                "run without the aid of the standard startup scripts or the scripts have been edited. If this was " +
-                "intentional, and you are attempting to use Java 11+ you may need to add the --add-exports and " +
-                "--add-opens jvm options from either jvm11-server.options or jvm11-client.options", e);
-            throw new RuntimeException(e);  // causes ExceptionInInitializerError, will prevent startup
-        }
-        catch (Throwable t)
-        {
-            logger.error("FATAL: Cannot initialize optimized memory deallocator.", t);
-            JVMStabilityInspector.inspectThrowable(t);
-            throw new RuntimeException(t); // causes ExceptionInInitializerError, will prevent startup
-        }
-    }
-
     private static final File tempDir = new File(JAVA_IO_TMPDIR.getString());
     private static final AtomicLong tempFileNum = new AtomicLong();
 
@@ -358,34 +321,6 @@ public final class FileUtils
         return folder.isAncestorOf(file);
     }
 
-    public static void clean(ByteBuffer buffer)
-    {
-        if (buffer == null || !buffer.isDirect())
-            return;
-
-        // TODO Once we can get rid of Java 8, it's simpler to call sun.misc.Unsafe.invokeCleaner(ByteBuffer),
-        // but need to take care of the attachment handling (i.e. whether 'buf' is a duplicate or slice) - that
-        // is different in sun.misc.Unsafe.invokeCleaner and this implementation.
-
-        try
-        {
-            Object cleaner = mhDirectBufferCleaner.bindTo(buffer).invoke();
-            if (cleaner != null)
-            {
-                // ((DirectBuffer) buf).cleaner().clean();
-                mhCleanerClean.bindTo(cleaner).invoke();
-            }
-        }
-        catch (RuntimeException e)
-        {
-            throw e;
-        }
-        catch (Throwable e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
     public static long parseFileSize(String value)
     {
         long result;
diff --git a/src/java/org/apache/cassandra/io/util/MmappedRegions.java b/src/java/org/apache/cassandra/io/util/MmappedRegions.java
index 578217e279..9852e78df3 100644
--- a/src/java/org/apache/cassandra/io/util/MmappedRegions.java
+++ b/src/java/org/apache/cassandra/io/util/MmappedRegions.java
@@ -32,6 +32,7 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.concurrent.RefCounted;
 import org.apache.cassandra.utils.concurrent.SharedCloseableImpl;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static java.util.stream.Stream.of;
 import static org.apache.cassandra.utils.Throwables.perform;
@@ -400,7 +401,7 @@ public class MmappedRegions extends SharedCloseableImpl
                                 () ->
                                 {
                                     if (buffer != null)
-                                        FileUtils.clean(buffer);
+                                        MemoryUtil.clean(buffer);
                                 }));
         }
     }
diff --git a/src/java/org/apache/cassandra/io/util/NIODataInputStream.java b/src/java/org/apache/cassandra/io/util/NIODataInputStream.java
index c75d44f198..43407ecbec 100644
--- a/src/java/org/apache/cassandra/io/util/NIODataInputStream.java
+++ b/src/java/org/apache/cassandra/io/util/NIODataInputStream.java
@@ -24,6 +24,8 @@ import java.nio.channels.SeekableByteChannel;
 
 import com.google.common.base.Preconditions;
 
+import org.apache.cassandra.utils.memory.MemoryUtil;
+
 /**
  * Rough equivalent of BufferedInputStream and DataInputStream wrapping the input stream of a File or Socket
  * Created to work around the fact that when BIS + DIS delegate to NIO for socket IO they will allocate large
@@ -73,7 +75,7 @@ public class NIODataInputStream extends RebufferingInputStream
     {
         channel.close();
         super.close();
-        FileUtils.clean(buffer);
+        MemoryUtil.clean(buffer);
         buffer = null;
     }
 
diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java
index c3a90732ee..e76cc234ed 100644
--- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java
+++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.utils.SyncUtil;
 import org.apache.cassandra.utils.concurrent.Transactional;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.utils.Throwables.merge;
 
@@ -77,7 +78,10 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
 
             if (buffer != null)
             {
-                try { FileUtils.clean(buffer); }
+                try
+                {
+                    MemoryUtil.clean(buffer);
+                }
                 catch (Throwable t) { accumulate = merge(accumulate, t); }
                 buffer = null;
             }
diff --git a/src/java/org/apache/cassandra/io/util/SimpleCachedBufferPool.java b/src/java/org/apache/cassandra/io/util/SimpleCachedBufferPool.java
index b7550fa69a..c4a639ae0c 100644
--- a/src/java/org/apache/cassandra/io/util/SimpleCachedBufferPool.java
+++ b/src/java/org/apache/cassandra/io/util/SimpleCachedBufferPool.java
@@ -23,6 +23,7 @@ import java.util.Queue;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.jctools.queues.MpmcArrayQueue;
 
 /**
@@ -92,7 +93,7 @@ public class SimpleCachedBufferPool
         // We use a bounded queue. By consequence if we have reached the maximum size for the buffer pool
         // offer will return false and we know that we can simply get rid of the buffer.
         if (!bufferPool.offer(buffer))
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
     }
 
     /**
@@ -103,7 +104,7 @@ public class SimpleCachedBufferPool
         ByteBuffer buffer = bufferPool.poll();
         while(buffer != null)
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             buffer = bufferPool.poll();
         }
     }
diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java b/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java
index 392edbadd6..3b3c701d34 100644
--- a/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java
+++ b/src/java/org/apache/cassandra/io/util/ThreadLocalByteBufferHolder.java
@@ -24,6 +24,7 @@ import java.util.EnumMap;
 import io.netty.util.concurrent.FastThreadLocal;
 
 import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 /**
  * Utility class that allow buffers to be reused by storing them in a thread local instance.
@@ -73,7 +74,7 @@ public final class ThreadLocalByteBufferHolder
         ByteBuffer buffer = reusableBB.get();
         if (buffer.capacity() < size)
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             buffer = bufferType.allocate(size);
             reusableBB.set(buffer);
         }
diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
index bc92407bef..c7963e431a 100644
--- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
+++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
@@ -25,6 +25,7 @@ import java.util.Map;
 import io.netty.util.concurrent.FastThreadLocal;
 import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 public final class ThreadLocalReadAheadBuffer
 {
@@ -144,7 +145,7 @@ public final class ThreadLocalReadAheadBuffer
         blockBuffer.clear();
         if (deallocate)
         {
-            FileUtils.clean(blockBuffer);
+            MemoryUtil.clean(blockBuffer);
             block.buffer = null;
         }
     }
diff --git a/src/java/org/apache/cassandra/journal/ActiveSegment.java b/src/java/org/apache/cassandra/journal/ActiveSegment.java
index fb5245de9b..ac05f78f2c 100644
--- a/src/java/org/apache/cassandra/journal/ActiveSegment.java
+++ b/src/java/org/apache/cassandra/journal/ActiveSegment.java
@@ -28,11 +28,11 @@ import java.util.concurrent.locks.LockSupport;
 
 import com.codahale.metrics.Timer;
 import org.apache.cassandra.db.TypeSizes;
-import org.apache.cassandra.io.util.*;
 import org.apache.cassandra.utils.*;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.concurrent.WaitQueue;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.utils.Simulate.With.MONITORS;
 
@@ -235,7 +235,7 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
         @Override
         void onUnreferenced()
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             try
             {
                 channel.close();
diff --git a/src/java/org/apache/cassandra/journal/OnDiskIndex.java b/src/java/org/apache/cassandra/journal/OnDiskIndex.java
index 8c662d889a..d82e5df5f3 100644
--- a/src/java/org/apache/cassandra/journal/OnDiskIndex.java
+++ b/src/java/org/apache/cassandra/journal/OnDiskIndex.java
@@ -33,6 +33,7 @@ import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.AbstractIterator;
 import org.apache.cassandra.utils.Crc;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.journal.Journal.validateCRC;
 import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
@@ -97,7 +98,7 @@ final class OnDiskIndex<K> extends Index<K>
         }
         catch (Throwable e)
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             FileUtils.closeQuietly(channel);
             throw new JournalReadError(descriptor, file, e);
         }
@@ -117,7 +118,7 @@ final class OnDiskIndex<K> extends Index<K>
     {
         try
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             buffer = null;
             channel.close();
         }
diff --git a/src/java/org/apache/cassandra/journal/StaticSegment.java b/src/java/org/apache/cassandra/journal/StaticSegment.java
index 92dd999c2d..2dffd2bcb7 100644
--- a/src/java/org/apache/cassandra/journal/StaticSegment.java
+++ b/src/java/org/apache/cassandra/journal/StaticSegment.java
@@ -35,6 +35,7 @@ import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.Closeable;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.concurrent.Ref;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 /**
  * An immutable data segment that is no longer written to.
@@ -207,7 +208,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
         @Override
         void onUnreferenced()
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
             FileUtils.closeQuietly(channel);
             index.close();
             if (discard)
@@ -335,7 +336,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
         public void close()
         {
             FileUtils.closeQuietly(channel);
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
         }
 
         public abstract boolean advance();
diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
index 6bdea57d68..57d824885a 100644
--- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
+++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
@@ -58,7 +58,7 @@ import org.apache.cassandra.io.UnversionedSerializer;
 import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 /**
  * Utility methods to make ByteBuffers less painful
@@ -904,7 +904,7 @@ public class ByteBufferUtil
         {
             if (!allowBufferResize)
                 throw new IllegalStateException(String.format("output buffer is not large enough for data: current capacity %d, required %d", buf.capacity(), outputLength));
-            FileUtils.clean(buf);
+            MemoryUtil.clean(buf);
             buf = bufferType.allocate(outputLength);
         }
         else
diff --git a/src/java/org/apache/cassandra/utils/MerkleTree.java b/src/java/org/apache/cassandra/utils/MerkleTree.java
index cadc47ed27..e41cd1dff2 100644
--- a/src/java/org/apache/cassandra/utils/MerkleTree.java
+++ b/src/java/org/apache/cassandra/utils/MerkleTree.java
@@ -40,7 +40,6 @@ import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.memory.MemoryUtil;
 
@@ -987,7 +986,7 @@ public class MerkleTree
             Object attachment = MemoryUtil.getAttachment(buffer);
             if (attachment instanceof Ref.DirectBufferRef)
                 ((Ref.DirectBufferRef) attachment).release();
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
         }
 
         abstract int hashBytesOffset();
diff --git a/src/java/org/apache/cassandra/utils/memory/BufferPool.java b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
index 90c95228f9..1e00e49611 100644
--- a/src/java/org/apache/cassandra/utils/memory/BufferPool.java
+++ b/src/java/org/apache/cassandra/utils/memory/BufferPool.java
@@ -46,7 +46,6 @@ import org.slf4j.LoggerFactory;
 import io.netty.util.concurrent.FastThreadLocal;
 
 import org.apache.cassandra.io.compress.BufferType;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.metrics.BufferPoolMetrics;
 import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.Shared;
@@ -816,7 +815,7 @@ public class BufferPool
 
             if (chunk == null)
             {
-                FileUtils.clean(buffer);
+                MemoryUtil.clean(buffer);
                 updateOverflowMemoryUsage(-size);
             }
             else
@@ -1563,7 +1562,7 @@ public class BufferPool
             if (parent != null)
                 parent.free(slab);
             else
-                FileUtils.clean(slab);
+                MemoryUtil.clean(slab);
         }
 
         static void unsafeRecycle(Chunk chunk)
diff --git a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java
index 724c673f88..ec94da2c93 100644
--- a/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java
+++ b/src/java/org/apache/cassandra/utils/memory/MemoryUtil.java
@@ -25,6 +25,7 @@ import java.nio.ByteOrder;
 import com.sun.jna.Native;
 
 import sun.misc.Unsafe;
+import sun.nio.ch.DirectBuffer;
 
 public abstract class MemoryUtil
 {
@@ -322,4 +323,16 @@ public abstract class MemoryUtil
     {
         getBytes(sourceAddress, targetBuffer, 0, length);
     }
+
+    public static void clean(ByteBuffer buffer)
+    {
+        if (buffer == null || !buffer.isDirect())
+            return;
+
+        DirectBuffer db = (DirectBuffer) buffer;
+        if (db.attachment() != null)
+            return; // duplicate or slice
+
+        unsafe.invokeCleaner(buffer);
+    }
 }
diff --git a/src/java/org/apache/cassandra/utils/memory/SlabAllocator.java b/src/java/org/apache/cassandra/utils/memory/SlabAllocator.java
index 441cb5b040..9c0b263f10 100644
--- a/src/java/org/apache/cassandra/utils/memory/SlabAllocator.java
+++ b/src/java/org/apache/cassandra/utils/memory/SlabAllocator.java
@@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
@@ -116,7 +115,7 @@ public class SlabAllocator extends MemtableBufferAllocator
     public void setDiscarded()
     {
         for (Region region : offHeapRegions)
-            FileUtils.clean(region.data);
+            MemoryUtil.clean(region.data);
         super.setDiscarded();
     }
 
diff --git a/test/long/org/apache/cassandra/hints/HintsWriteThenReadTest.java b/test/long/org/apache/cassandra/hints/HintsWriteThenReadTest.java
index 1b78c48b6e..7a20787889 100644
--- a/test/long/org/apache/cassandra/hints/HintsWriteThenReadTest.java
+++ b/test/long/org/apache/cassandra/hints/HintsWriteThenReadTest.java
@@ -36,11 +36,11 @@ import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.RowUpdateBuilder;
 import org.apache.cassandra.db.rows.Cell;
 import org.apache.cassandra.db.rows.Row;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 import static org.apache.cassandra.Util.dk;
 import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
@@ -152,7 +152,7 @@ public class HintsWriteThenReadTest
         {
             write(session, timestamp);
         }
-        FileUtils.clean(buffer);
+        MemoryUtil.clean(buffer);
     }
 
     private void write(HintsWriter.Session session, long timestamp) throws IOException
diff --git a/test/unit/org/apache/cassandra/db/MmapFileTest.java b/test/unit/org/apache/cassandra/db/MmapFileTest.java
index 4619df3b7d..c3b9482db8 100644
--- a/test/unit/org/apache/cassandra/db/MmapFileTest.java
+++ b/test/unit/org/apache/cassandra/db/MmapFileTest.java
@@ -32,6 +32,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 public class MmapFileTest
 {
@@ -86,8 +87,8 @@ public class MmapFileTest
                 buffer.putInt(42);
                 buffer.putInt(42);
                 buffer.putInt(42);
-                
-                FileUtils.clean(buffer);
+
+                MemoryUtil.clean(buffer);
             }
 
             mmapCount = (Long) mbs.getAttribute(bpmName, "Count");
@@ -114,7 +115,7 @@ public class MmapFileTest
                 buffer.putInt(42);
                 buffer.putInt(42);
 
-                FileUtils.clean(buffer);
+                MemoryUtil.clean(buffer);
             }
 
             mmapCount = (Long) mbs.getAttribute(bpmName, "Count");
@@ -139,7 +140,7 @@ public class MmapFileTest
                 buffer.putInt(42);
                 buffer.putInt(42);
 
-                FileUtils.clean(buffer);
+                MemoryUtil.clean(buffer);
             }
 
             mmapCount = (Long) mbs.getAttribute(bpmName, "Count");
diff --git a/test/unit/org/apache/cassandra/hints/HintsMaker.java b/test/unit/org/apache/cassandra/hints/HintsMaker.java
index f2eb5f4cb4..a89baa6a02 100644
--- a/test/unit/org/apache/cassandra/hints/HintsMaker.java
+++ b/test/unit/org/apache/cassandra/hints/HintsMaker.java
@@ -44,11 +44,11 @@ import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.util.File;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 public class HintsMaker
 {
@@ -202,7 +202,7 @@ public class HintsMaker
             }
             finally
             {
-                FileUtils.clean(buffer);
+                MemoryUtil.clean(buffer);
             }
         }
 
diff --git a/test/unit/org/apache/cassandra/hints/HintsReaderTest.java b/test/unit/org/apache/cassandra/hints/HintsReaderTest.java
index 94be7b2b8e..a47d0f4ab8 100644
--- a/test/unit/org/apache/cassandra/hints/HintsReaderTest.java
+++ b/test/unit/org/apache/cassandra/hints/HintsReaderTest.java
@@ -45,11 +45,11 @@ import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.exceptions.UnknownTableException;
 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.util.DataInputBuffer;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaTestUtil;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.hamcrest.Matchers;
 
 import static org.junit.Assert.assertEquals;
@@ -99,7 +99,7 @@ public class HintsReaderTest
                     session.append(Hint.create(m, timestamp));
                 }
             }
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
         }
 
         Assert.assertThat(descriptor.hintsFileSize(directory), Matchers.greaterThan(0L));
diff --git a/test/unit/org/apache/cassandra/hints/HintsStoreTest.java b/test/unit/org/apache/cassandra/hints/HintsStoreTest.java
index 624783d387..fdb2206cc9 100644
--- a/test/unit/org/apache/cassandra/hints/HintsStoreTest.java
+++ b/test/unit/org/apache/cassandra/hints/HintsStoreTest.java
@@ -41,8 +41,8 @@ import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.RowUpdateBuilder;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.hamcrest.Matchers;
 
 import static org.junit.Assert.assertEquals;
@@ -176,7 +176,7 @@ public class HintsStoreTest
                 for (int i = 0; i < hintsCount; i++)
                     session.append(createHint(i, hintCreationTime));
             }
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
         }
         Assert.assertThat(descriptor.hintsFileSize(directory), Matchers.greaterThan(0L));
         return new File(directory, descriptor.fileName()).lastModified(); // hint file last modified time
diff --git a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
index af4b458fec..9a7473873e 100644
--- a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
@@ -27,6 +27,7 @@ import org.apache.cassandra.io.compress.CompressionMetadata;
 import org.apache.cassandra.io.filesystem.ListenableFileSystem;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.assertj.core.api.Assertions;
 
 import org.junit.Assert;
@@ -107,7 +108,7 @@ public class CompressedChunkReaderTest
         }
         finally
         {
-            FileUtils.clean(buffer);
+            MemoryUtil.clean(buffer);
         }}
 
     private static Gen<SequentialWriterOption> options()
diff --git a/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java b/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java
index dab600111e..7a5e82c917 100644
--- a/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java
+++ b/test/unit/org/apache/cassandra/streaming/async/StreamCompressionSerializerTest.java
@@ -37,8 +37,8 @@ import net.jpountz.lz4.LZ4Factory;
 import net.jpountz.lz4.LZ4SafeDecompressor;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.util.DataInputBuffer;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 
 public class StreamCompressionSerializerTest
 {
@@ -64,9 +64,9 @@ public class StreamCompressionSerializerTest
     public void tearDown()
     {
         if (input != null)
-            FileUtils.clean(input);
+            MemoryUtil.clean(input);
         if (compressed != null)
-            FileUtils.clean(compressed);
+            MemoryUtil.clean(compressed);
         if (output != null && output.refCnt() > 0)
             output.release(output.refCnt());
     }
@@ -104,7 +104,7 @@ public class StreamCompressionSerializerTest
         StreamCompressionSerializer.serialize(compressor, input, VERSION)
                                    .write(size -> {
                                        if (compressed != null)
-                                           FileUtils.clean(compressed);
+                                           MemoryUtil.clean(compressed);
                                        return compressed = ByteBuffer.allocateDirect(size);
                                    });
         input.flip();
diff --git a/test/unit/org/apache/cassandra/transport/WriteBytesTest.java b/test/unit/org/apache/cassandra/transport/WriteBytesTest.java
index 1dd1c796e8..f28ff4854f 100644
--- a/test/unit/org/apache/cassandra/transport/WriteBytesTest.java
+++ b/test/unit/org/apache/cassandra/transport/WriteBytesTest.java
@@ -22,8 +22,8 @@ import org.junit.Test;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.Generators;
+import org.apache.cassandra.utils.memory.MemoryUtil;
 import org.assertj.core.api.Assertions;
 
 import static org.quicktheories.QuickTheory.qt;
@@ -51,7 +51,7 @@ public class WriteBytesTest
             Assertions.assertThat(buf.writerIndex()).isEqualTo(size);
             for (int i = 0; i < size; i++)
                 Assertions.assertThat(buf.getByte(buf.readerIndex() + i)).describedAs("byte mismatch at index %d", i).isEqualTo(bb.get(bb.position() + i));
-            FileUtils.clean(bb);
+            MemoryUtil.clean(bb);
         });
     }
 
diff --git a/test/unit/org/apache/cassandra/utils/memory/MemoryUtilTest.java b/test/unit/org/apache/cassandra/utils/memory/MemoryUtilTest.java
new file mode 100644
index 0000000000..392d3d5e06
--- /dev/null
+++ b/test/unit/org/apache/cassandra/utils/memory/MemoryUtilTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.utils.memory;
+
+import java.lang.management.BufferPoolMXBean;
+import java.lang.management.ManagementFactory;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.assertj.core.api.Assertions;
+
+public class MemoryUtilTest
+{
+
+    @Test
+    public void testClean()
+    {
+        // We assert direct pool state changes as no ByteBuffer state changes are observable as a result of cleaner
+        // executing.
+        BufferPoolMXBean directPool = getDirectBufferPool();
+
+        int bufferSize = 1024 * 1024; // some non-insignificant size
+        ByteBuffer original = ByteBuffer.allocateDirect(bufferSize);
+
+        long memoryUsedBefore = directPool.getMemoryUsed();
+        MemoryUtil.clean(original);
+        long memoryUsedAfter = directPool.getMemoryUsed();
+
+        Assert.assertEquals("Direct memory used should decrease by buffer capacity",
+                            // Allow 5% tolerance for other activities
+                            memoryUsedBefore - bufferSize, memoryUsedAfter, bufferSize * 0.05);
+    }
+
+    @Test
+    public void testCleanViewDoesNotThrow()
+    {
+        // Use a large buffer to likely get mmap'd memory from the OS. This ensures that if cleaning a view incorrectly
+        // unmaps the original buffer's memory, subsequent access to 'original' would more reliably fail.
+        // For context: glibc's mmap threshold is 32MB on 64-bit systems
+        ByteBuffer original = ByteBuffer.allocateDirect(64 * 1024 * 1024);
+
+        ByteBuffer slice = original.slice();
+        MemoryUtil.clean(slice);
+        try
+        {
+            original.putInt(10);
+        }
+        catch (Exception exc)
+        {
+            Assertions.fail("Unable to write to original buffer after cleaning (slice). " + exc.getMessage(), exc);
+        }
+
+        ByteBuffer duplicate = original.duplicate();
+        MemoryUtil.clean(duplicate);
+
+        try
+        {
+            original.putInt(10);
+        }
+        catch (Exception exc)
+        {
+            Assertions.fail("Unable to write to original buffer after cleaning (duplicate). " + exc.getMessage(), exc);
+        }
+    }
+
+    @Test
+    public void testCleanNonDirectDoesNotThrow()
+    {
+        ByteBuffer original = ByteBuffer.allocate(16);
+        MemoryUtil.clean(original);
+    }
+
+    private static BufferPoolMXBean getDirectBufferPool()
+    {
+        List<BufferPoolMXBean> pools = ManagementFactory.getPlatformMXBeans(BufferPoolMXBean.class);
+        for (BufferPoolMXBean pool : pools)
+        {
+            if (pool.getName().equals("direct"))
+            {
+                return pool;
+            }
+        }
+        throw new IllegalArgumentException("Direct buffer pool not found");
+    }
+}
\ No newline at end of file