diff --git a/CHANGES.txt b/CHANGES.txt
index dd3e503acb..0dbc28102a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -100,6 +100,7 @@
 
 
 2.1.6
+ * Fix for harmless exceptions logged as ERROR (CASSANDRA-8564)
  * Delete processed sstables in sstablesplit/sstableupgrade (CASSANDRA-8606)
  * Improve sstable exclusion from partition tombstones (CASSANDRA-9298)
  * Validate the indexed column rather than the cell's contents for 2i (CASSANDRA-9057)
diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutor.java
index a301923503..7226db8f7c 100644
--- a/src/java/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutor.java
+++ b/src/java/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutor.java
@@ -19,6 +19,10 @@ package org.apache.cassandra.concurrent;
 
 import java.util.concurrent.*;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 
 /**
@@ -30,19 +34,42 @@ import org.apache.cassandra.utils.JVMStabilityInspector;
  */
 public class DebuggableScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor
 {
+    private static final Logger logger = LoggerFactory.getLogger(DebuggableScheduledThreadPoolExecutor.class);
+
+    public static final RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler()
+    {
+        public void rejectedExecution(Runnable task, ThreadPoolExecutor executor)
+        {
+            if (executor.isShutdown())
+            {
+                if (!StorageService.instance.isInShutdownHook())
+                    throw new RejectedExecutionException("ScheduledThreadPoolExecutor has shut down.");
+
+                logger.debug("ScheduledThreadPoolExecutor has shut down as part of C* shutdown");
+            }
+            else
+            {
+                throw new AssertionError("Unknown rejection of ScheduledThreadPoolExecutor task");
+            }
+        }
+    };
+
     public DebuggableScheduledThreadPoolExecutor(int corePoolSize, String threadPoolName, int priority)
     {
         super(corePoolSize, new NamedThreadFactory(threadPoolName, priority));
+        setRejectedExecutionHandler(rejectedExecutionHandler);
     }
 
     public DebuggableScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory)
     {
         super(corePoolSize, threadFactory);
+        setRejectedExecutionHandler(rejectedExecutionHandler);
     }
 
     public DebuggableScheduledThreadPoolExecutor(String threadPoolName)
     {
         this(1, threadPoolName, Thread.NORM_PRIORITY);
+        setRejectedExecutionHandler(rejectedExecutionHandler);
     }
 
     // We need this as well as the wrapper for the benefit of non-repeating tasks
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index c17d2d74e8..f03870b84a 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -204,9 +204,15 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public volatile VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(getPartitioner());
 
     private Thread drainOnShutdown = null;
+    private boolean inShutdownHook = false;
 
     public static final StorageService instance = new StorageService();
 
+    public boolean isInShutdownHook()
+    {
+        return inShutdownHook;
+    }
+
     public static IPartitioner getPartitioner()
     {
         return DatabaseDescriptor.getPartitioner();
@@ -613,6 +619,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
             @Override
             public void runMayThrow() throws InterruptedException
             {
+                inShutdownHook = true;
                 ExecutorService counterMutationStage = StageManager.getStage(Stage.COUNTER_MUTATION);
                 ExecutorService mutationStage = StageManager.getStage(Stage.MUTATION);
                 if (mutationStage.isShutdown() && counterMutationStage.isShutdown())