diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
index 25859d4a7169..6ae9bbe113f4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -216,9 +216,7 @@ public class BucketCache implements BlockCache, HeapSize {
   });
 
   /** Statistics thread schedule pool (for heavy debugging, could remove) */
-  private transient final ScheduledExecutorService scheduleThreadPool =
-    Executors.newScheduledThreadPool(1,
-      new ThreadFactoryBuilder().setNameFormat("BucketCacheStatsExecutor").setDaemon(true).build());
+  private final transient ScheduledExecutorService scheduleThreadPool;
 
   // Allocate or free space for the block
   private transient BucketAllocator bucketAllocator;
@@ -244,6 +242,8 @@ public class BucketCache implements BlockCache, HeapSize {
   private static final String FILE_VERIFY_ALGORITHM =
     "hbase.bucketcache.persistent.file.integrity.check.algorithm";
   private static final String DEFAULT_FILE_VERIFY_ALGORITHM = "MD5";
+  private static final String STAT_THREAD_ENABLE_KEY = "hbase.bucketcache.stat.enable";
+  private static final boolean STAT_THREAD_ENABLE_DEFAULT = false;
 
   /**
    * Use {@link java.security.MessageDigest} class's encryption algorithms to check
@@ -325,11 +325,16 @@ public BucketCache(String ioEngineName, long capacity, int blockSize, int[] buck
     }
     startWriterThreads();
 
-    // Run the statistics thread periodically to print the cache statistics log
-    // TODO: Add means of turning this off.  Bit obnoxious running thread just to make a log
-    // every five minutes.
-    this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
+    // Run the statistics thread periodically to print the cache statistics log if it is enabled.
+    if (conf.getBoolean(STAT_THREAD_ENABLE_KEY, STAT_THREAD_ENABLE_DEFAULT)) {
+      this.scheduleThreadPool = Executors.newScheduledThreadPool(1,
+        new ThreadFactoryBuilder().setNameFormat("BucketCacheStatsExecutor").
+          setDaemon(true).build());
+      this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
         statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
+    } else {
+      this.scheduleThreadPool = null;
+    }
     LOG.info("Started bucket cache; ioengine=" + ioEngineName +
         ", capacity=" + StringUtils.byteDesc(capacity) +
       ", blockSize=" + StringUtils.byteDesc(blockSize) + ", writerThreadNum=" +
@@ -1316,7 +1321,9 @@ private void disableCache() {
     if (!cacheEnabled) return;
     cacheEnabled = false;
     ioEngine.shutdown();
-    this.scheduleThreadPool.shutdown();
+    if (this.scheduleThreadPool != null) {
+      this.scheduleThreadPool.shutdown();
+    }
     for (int i = 0; i < writerThreads.length; ++i) writerThreads[i].interrupt();
     this.ramCache.clear();
     if (!ioEngine.isPersistent() || persistencePath == null) {