/*
 * 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.hadoop.hbase.regionserver;

import static org.apache.hadoop.hbase.HConstants.HFILE_BLOCK_CACHE_MEMORY_SIZE_KEY;
import static org.apache.hadoop.hbase.HConstants.HFILE_BLOCK_CACHE_SIZE_KEY;
import static org.apache.hadoop.hbase.io.util.MemorySizeUtil.MEMSTORE_MEMORY_SIZE_KEY;
import static org.apache.hadoop.hbase.io.util.MemorySizeUtil.MEMSTORE_SIZE_KEY;

import java.lang.management.MemoryUsage;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ChoreService;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.ScheduledChore;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.io.hfile.BlockCache;
import org.apache.hadoop.hbase.io.hfile.CombinedBlockCache;
import org.apache.hadoop.hbase.io.hfile.ResizableBlockCache;
import org.apache.hadoop.hbase.io.util.MemorySizeUtil;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Manages tuning of Heap memory using <code>HeapMemoryTuner</code>. Most part of the heap memory is
 * split between Memstores and BlockCache. This manager helps in tuning sizes of both these
 * dynamically, as per the R/W load on the servers.
 */
@InterfaceAudience.Private
public class HeapMemoryManager {
  private static final Logger LOG = LoggerFactory.getLogger(HeapMemoryManager.class);
  private static final int CONVERT_TO_PERCENTAGE = 100;

  public static final String BLOCK_CACHE_SIZE_MAX_RANGE_KEY = "hfile.block.cache.size.max.range";
  public static final String BLOCK_CACHE_SIZE_MIN_RANGE_KEY = "hfile.block.cache.size.min.range";
  public static final String MEMSTORE_SIZE_MAX_RANGE_KEY =
    "hbase.regionserver.global.memstore.size.max.range";
  public static final String MEMSTORE_SIZE_MIN_RANGE_KEY =
    "hbase.regionserver.global.memstore.size.min.range";
  public static final String HBASE_RS_HEAP_MEMORY_TUNER_PERIOD =
    "hbase.regionserver.heapmemory.tuner.period";
  public static final int HBASE_RS_HEAP_MEMORY_TUNER_DEFAULT_PERIOD = 60 * 1000;
  public static final String HBASE_RS_HEAP_MEMORY_TUNER_CLASS =
    "hbase.regionserver.heapmemory.tuner.class";

  public static final float HEAP_OCCUPANCY_ERROR_VALUE = -0.0f;

  private float globalMemStorePercent;
  private float globalMemStorePercentMinRange;
  private float globalMemStorePercentMaxRange;

  private float blockCachePercent;
  private float blockCachePercentMinRange;
  private float blockCachePercentMaxRange;

  private float heapOccupancyPercent;

  private final ResizableBlockCache blockCache;
  // TODO : remove this and mark regionServerAccounting as the observer directly
  private final FlushRequester memStoreFlusher;
  private final Server server;
  private final RegionServerAccounting regionServerAccounting;

  private HeapMemoryTunerChore heapMemTunerChore = null;
  private final boolean tunerOn;
  private final int defaultChorePeriod;
  private final float heapOccupancyLowWatermark;
  private final float minFreeHeapFraction;

  private final long maxHeapSize;
  {
    // note that this initialization still isn't threadsafe, because updating a long isn't atomic.
    long tempMaxHeap = -1L;
    try {
      final MemoryUsage usage = MemorySizeUtil.safeGetHeapMemoryUsage();
      if (usage != null) {
        tempMaxHeap = usage.getMax();
      }
    } finally {
      maxHeapSize = tempMaxHeap;
    }
  }

  private MetricsHeapMemoryManager metricsHeapMemoryManager;

  private List<HeapMemoryTuneObserver> tuneObservers = new ArrayList<>();

  HeapMemoryManager(BlockCache blockCache, FlushRequester memStoreFlusher, Server server,
    RegionServerAccounting regionServerAccounting) {
    Configuration conf = server.getConfiguration();
    this.blockCache = toResizableBlockCache(blockCache);
    this.memStoreFlusher = memStoreFlusher;
    this.server = server;
    this.regionServerAccounting = regionServerAccounting;
    this.minFreeHeapFraction = MemorySizeUtil.getRegionServerMinFreeHeapFraction(conf);
    this.tunerOn = doInit(conf);
    this.defaultChorePeriod =
      conf.getInt(HBASE_RS_HEAP_MEMORY_TUNER_PERIOD, HBASE_RS_HEAP_MEMORY_TUNER_DEFAULT_PERIOD);
    this.heapOccupancyLowWatermark = conf.getFloat(HConstants.HEAP_OCCUPANCY_LOW_WATERMARK_KEY,
      HConstants.DEFAULT_HEAP_OCCUPANCY_LOW_WATERMARK);
    metricsHeapMemoryManager = new MetricsHeapMemoryManager();
  }

  private ResizableBlockCache toResizableBlockCache(BlockCache blockCache) {
    if (blockCache instanceof CombinedBlockCache) {
      return ((CombinedBlockCache) blockCache).getFirstLevelCache();
    } else {
      return (ResizableBlockCache) blockCache;
    }
  }

  private boolean doInit(Configuration conf) {
    boolean tuningEnabled = true;
    globalMemStorePercent = MemorySizeUtil.getGlobalMemStoreHeapPercent(conf, false);
    blockCachePercent = MemorySizeUtil.getBlockCacheHeapPercent(conf);
    MemorySizeUtil.validateRegionServerHeapMemoryAllocation(conf);
    // Initialize max and min range for memstore heap space
    globalMemStorePercentMinRange =
      conf.getFloat(MEMSTORE_SIZE_MIN_RANGE_KEY, globalMemStorePercent);
    globalMemStorePercentMaxRange =
      conf.getFloat(MEMSTORE_SIZE_MAX_RANGE_KEY, globalMemStorePercent);
    if (globalMemStorePercent < globalMemStorePercentMinRange) {
      LOG.warn(
        "Setting {} to {} (lookup order: {} -> {}), "
          + "because supplied value greater than initial memstore size.",
        MEMSTORE_SIZE_MIN_RANGE_KEY, globalMemStorePercent, MEMSTORE_MEMORY_SIZE_KEY,
        MEMSTORE_SIZE_KEY);
      globalMemStorePercentMinRange = globalMemStorePercent;
      conf.setFloat(MEMSTORE_SIZE_MIN_RANGE_KEY, globalMemStorePercentMinRange);
    }
    if (globalMemStorePercent > globalMemStorePercentMaxRange) {
      LOG.warn(
        "Setting {} to {} (lookup order: {} -> {}), "
          + "because supplied value less than initial memstore size.",
        MEMSTORE_SIZE_MAX_RANGE_KEY, globalMemStorePercent, MEMSTORE_MEMORY_SIZE_KEY,
        MEMSTORE_SIZE_KEY);
      globalMemStorePercentMaxRange = globalMemStorePercent;
      conf.setFloat(MEMSTORE_SIZE_MAX_RANGE_KEY, globalMemStorePercentMaxRange);
    }
    if (
      globalMemStorePercent == globalMemStorePercentMinRange
        && globalMemStorePercent == globalMemStorePercentMaxRange
    ) {
      tuningEnabled = false;
    }
    // Initialize max and min range for block cache
    blockCachePercentMinRange = conf.getFloat(BLOCK_CACHE_SIZE_MIN_RANGE_KEY, blockCachePercent);
    blockCachePercentMaxRange = conf.getFloat(BLOCK_CACHE_SIZE_MAX_RANGE_KEY, blockCachePercent);
    if (blockCachePercent < blockCachePercentMinRange) {
      LOG.warn(
        "Setting {} to {} (lookup order: {} -> {}), "
          + "because supplied value greater than initial block cache size.",
        BLOCK_CACHE_SIZE_MIN_RANGE_KEY, blockCachePercent, HFILE_BLOCK_CACHE_MEMORY_SIZE_KEY,
        HFILE_BLOCK_CACHE_SIZE_KEY);
      blockCachePercentMinRange = blockCachePercent;
      conf.setFloat(BLOCK_CACHE_SIZE_MIN_RANGE_KEY, blockCachePercentMinRange);
    }
    if (blockCachePercent > blockCachePercentMaxRange) {
      LOG.warn(
        "Setting {} to {} (lookup order: {} -> {}), "
          + "because supplied value less than initial block cache size.",
        BLOCK_CACHE_SIZE_MAX_RANGE_KEY, blockCachePercent, HFILE_BLOCK_CACHE_MEMORY_SIZE_KEY,
        HFILE_BLOCK_CACHE_SIZE_KEY);
      blockCachePercentMaxRange = blockCachePercent;
      conf.setFloat(BLOCK_CACHE_SIZE_MAX_RANGE_KEY, blockCachePercentMaxRange);
    }
    if (
      tuningEnabled && blockCachePercent == blockCachePercentMinRange
        && blockCachePercent == blockCachePercentMaxRange
    ) {
      tuningEnabled = false;
    }

    checkHeapMemoryLimits(MEMSTORE_SIZE_MAX_RANGE_KEY, globalMemStorePercentMaxRange,
      BLOCK_CACHE_SIZE_MIN_RANGE_KEY, blockCachePercentMinRange);
    checkHeapMemoryLimits(MEMSTORE_SIZE_MIN_RANGE_KEY, globalMemStorePercentMinRange,
      BLOCK_CACHE_SIZE_MAX_RANGE_KEY, blockCachePercentMaxRange);

    return tuningEnabled;
  }

  public void start(ChoreService service) {
    LOG.info("Starting, tuneOn={}", this.tunerOn);
    this.heapMemTunerChore = new HeapMemoryTunerChore();
    service.scheduleChore(heapMemTunerChore);
    if (tunerOn) {
      // Register HeapMemoryTuner as a memstore flush listener
      memStoreFlusher.registerFlushRequestListener(heapMemTunerChore);
    }
  }

  public void stop() {
    // The thread is Daemon. Just interrupting the ongoing process.
    LOG.info("Stopping");
    this.heapMemTunerChore.shutdown(true);
  }

  public void registerTuneObserver(HeapMemoryTuneObserver observer) {
    this.tuneObservers.add(observer);
  }

  // Used by the test cases.
  boolean isTunerOn() {
    return this.tunerOn;
  }

  /** Returns heap occupancy percentage, 0 &lt;= n &lt;= 1. or -0.0 for error asking JVM */
  public float getHeapOccupancyPercent() {
    return this.heapOccupancyPercent == Float.MAX_VALUE
      ? HEAP_OCCUPANCY_ERROR_VALUE
      : this.heapOccupancyPercent;
  }

  private boolean isHeapMemoryUsageExceedingLimit(float memStoreFraction,
    float blockCacheFraction) {
    // Use integer percentage to avoid subtle float precision issues and ensure consistent
    // comparison. This also maintains backward compatibility with previous logic relying on int
    // truncation.
    int memStorePercent = (int) (memStoreFraction * CONVERT_TO_PERCENTAGE);
    int blockCachePercent = (int) (blockCacheFraction * CONVERT_TO_PERCENTAGE);
    int minFreeHeapPercent = (int) (this.minFreeHeapFraction * CONVERT_TO_PERCENTAGE);

    return memStorePercent + blockCachePercent + minFreeHeapPercent > CONVERT_TO_PERCENTAGE;
  }

  private void checkHeapMemoryLimits(String memStoreConfKey, float memStoreFraction,
    String blockCacheConfKey, float blockCacheFraction) {
    if (isHeapMemoryUsageExceedingLimit(memStoreFraction, blockCacheFraction)) {
      throw new RuntimeException(String.format(
        "Current heap configuration for MemStore and BlockCache exceeds the allowed heap usage. "
          + "At least %.2f of the heap must remain free to ensure stable RegionServer operation. "
          + "Please check the settings for %s and %s in your configuration. "
          + "%s is %.2f and %s is %.2f",
        minFreeHeapFraction, memStoreConfKey, blockCacheConfKey, memStoreConfKey, memStoreFraction,
        blockCacheConfKey, blockCacheFraction));
    }
  }

  private class HeapMemoryTunerChore extends ScheduledChore implements FlushRequestListener {
    private HeapMemoryTuner heapMemTuner;
    private AtomicLong blockedFlushCount = new AtomicLong();
    private AtomicLong unblockedFlushCount = new AtomicLong();
    private long evictCount = 0L;
    private long cacheMissCount = 0L;
    private TunerContext tunerContext = new TunerContext();
    private boolean alarming = false;

    public HeapMemoryTunerChore() {
      super(server.getServerName() + "-HeapMemoryTunerChore", server, defaultChorePeriod);
      Class<? extends HeapMemoryTuner> tunerKlass = server.getConfiguration().getClass(
        HBASE_RS_HEAP_MEMORY_TUNER_CLASS, DefaultHeapMemoryTuner.class, HeapMemoryTuner.class);
      heapMemTuner = ReflectionUtils.newInstance(tunerKlass, server.getConfiguration());
      tunerContext.setOffheapMemStore(regionServerAccounting.isOffheap());
    }

    @Override
    protected void chore() {
      // Sample heap occupancy
      final MemoryUsage usage = MemorySizeUtil.safeGetHeapMemoryUsage();
      if (usage != null) {
        heapOccupancyPercent = (float) usage.getUsed() / (float) usage.getCommitted();
      } else {
        // previously, an exception would have meant death for the tuning chore
        // so switch to alarming so that we similarly stop tuning until we get
        // heap usage information again.
        heapOccupancyPercent = Float.MAX_VALUE;
      }
      // If we are above the heap occupancy alarm low watermark, switch to short
      // sleeps for close monitoring. Stop autotuning, we are in a danger zone.
      if (heapOccupancyPercent >= heapOccupancyLowWatermark) {
        if (!alarming) {
          LOG.warn("heapOccupancyPercent " + heapOccupancyPercent
            + " is above heap occupancy alarm watermark (" + heapOccupancyLowWatermark + ")");
          alarming = true;
        }
        metricsHeapMemoryManager.increaseAboveHeapOccupancyLowWatermarkCounter();
        triggerNow();
        try {
          // Need to sleep ourselves since we've told the chore's sleeper
          // to skip the next sleep cycle.
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          // Interrupted, propagate
          Thread.currentThread().interrupt();
        }
      } else {
        if (alarming) {
          LOG.info("heapOccupancyPercent " + heapOccupancyPercent
            + " is now below the heap occupancy alarm watermark (" + heapOccupancyLowWatermark
            + ")");
          alarming = false;
        }
      }
      // Autotune if tuning is enabled and allowed
      if (tunerOn && !alarming) {
        tune();
      }
    }

    private void tune() {
      // TODO check if we can increase the memory boundaries
      // while remaining in the limits
      long curEvictCount;
      long curCacheMisCount;
      long blockedFlushCnt;
      long unblockedFlushCnt;
      curEvictCount = blockCache.getStats().getEvictedCount();
      tunerContext.setEvictCount(curEvictCount - evictCount);
      evictCount = curEvictCount;
      curCacheMisCount = blockCache.getStats().getMissCachingCount();
      tunerContext.setCacheMissCount(curCacheMisCount - cacheMissCount);
      cacheMissCount = curCacheMisCount;
      blockedFlushCnt = blockedFlushCount.getAndSet(0);
      tunerContext.setBlockedFlushCount(blockedFlushCnt);
      metricsHeapMemoryManager.updateBlockedFlushCount(blockedFlushCnt);
      unblockedFlushCnt = unblockedFlushCount.getAndSet(0);
      tunerContext.setUnblockedFlushCount(unblockedFlushCnt);
      metricsHeapMemoryManager.updateUnblockedFlushCount(unblockedFlushCnt);
      tunerContext.setCurBlockCacheUsed((float) blockCache.getCurrentSize() / maxHeapSize);
      metricsHeapMemoryManager.setCurBlockCacheSizeGauge(blockCache.getCurrentSize());
      long globalMemstoreDataSize = regionServerAccounting.getGlobalMemStoreDataSize();
      long globalMemstoreHeapSize = regionServerAccounting.getGlobalMemStoreHeapSize();
      long globalMemStoreOffHeapSize = regionServerAccounting.getGlobalMemStoreOffHeapSize();
      tunerContext.setCurMemStoreUsed((float) globalMemstoreHeapSize / maxHeapSize);
      metricsHeapMemoryManager.setCurMemStoreSizeGauge(globalMemstoreDataSize);
      metricsHeapMemoryManager.setCurMemStoreOnHeapSizeGauge(globalMemstoreHeapSize);
      metricsHeapMemoryManager.setCurMemStoreOffHeapSizeGauge(globalMemStoreOffHeapSize);
      tunerContext.setCurBlockCacheSize(blockCachePercent);
      tunerContext.setCurMemStoreSize(globalMemStorePercent);
      TunerResult result = null;
      try {
        result = this.heapMemTuner.tune(tunerContext);
      } catch (Throwable t) {
        LOG.error("Exception thrown from the HeapMemoryTuner implementation", t);
      }
      if (result != null && result.needsTuning()) {
        float memstoreSize = result.getMemStoreSize();
        float blockCacheSize = result.getBlockCacheSize();
        LOG.debug("From HeapMemoryTuner new memstoreSize: " + memstoreSize
          + ". new blockCacheSize: " + blockCacheSize);
        if (memstoreSize < globalMemStorePercentMinRange) {
          LOG.info("New memstoreSize from HeapMemoryTuner " + memstoreSize + " is below min level "
            + globalMemStorePercentMinRange + ". Resetting memstoreSize to min size");
          memstoreSize = globalMemStorePercentMinRange;
        } else if (memstoreSize > globalMemStorePercentMaxRange) {
          LOG.info("New memstoreSize from HeapMemoryTuner " + memstoreSize + " is above max level "
            + globalMemStorePercentMaxRange + ". Resetting memstoreSize to max size");
          memstoreSize = globalMemStorePercentMaxRange;
        }
        if (blockCacheSize < blockCachePercentMinRange) {
          LOG.info(
            "New blockCacheSize from HeapMemoryTuner " + blockCacheSize + " is below min level "
              + blockCachePercentMinRange + ". Resetting blockCacheSize to min size");
          blockCacheSize = blockCachePercentMinRange;
        } else if (blockCacheSize > blockCachePercentMaxRange) {
          LOG.info(
            "New blockCacheSize from HeapMemoryTuner " + blockCacheSize + " is above max level "
              + blockCachePercentMaxRange + ". Resetting blockCacheSize to min size");
          blockCacheSize = blockCachePercentMaxRange;
        }

        if (isHeapMemoryUsageExceedingLimit(memstoreSize, blockCacheSize)) {
          LOG.info("Current heap configuration from HeapMemoryTuner exceeds "
            + "the allowed heap usage. At least " + minFreeHeapFraction
            + " of the heap must remain free to ensure stable RegionServer operation. "
            + MEMSTORE_SIZE_KEY + " is " + memstoreSize + " and " + HFILE_BLOCK_CACHE_SIZE_KEY
            + " is " + blockCacheSize);
          // NOTE: In the future, we might adjust values to not exceed limits,
          // but for now tuning is skipped if over threshold.
        } else {
          int memStoreDeltaSize =
            (int) ((memstoreSize - globalMemStorePercent) * CONVERT_TO_PERCENTAGE);
          int blockCacheDeltaSize =
            (int) ((blockCacheSize - blockCachePercent) * CONVERT_TO_PERCENTAGE);
          metricsHeapMemoryManager.updateMemStoreDeltaSizeHistogram(memStoreDeltaSize);
          metricsHeapMemoryManager.updateBlockCacheDeltaSizeHistogram(blockCacheDeltaSize);
          long newBlockCacheSize = (long) (maxHeapSize * blockCacheSize);
          // we could have got an increase or decrease in size for the offheap memstore
          // also if the flush had happened due to heap overhead. In that case it is ok
          // to adjust the onheap memstore limit configs
          long newMemstoreSize = (long) (maxHeapSize * memstoreSize);
          LOG.info("Setting block cache heap size to " + newBlockCacheSize
            + " and memstore heap size to " + newMemstoreSize);
          blockCachePercent = blockCacheSize;
          blockCache.setMaxSize(newBlockCacheSize);
          globalMemStorePercent = memstoreSize;
          // Internally sets it to RegionServerAccounting
          // TODO : Set directly on RSAccounting??
          memStoreFlusher.setGlobalMemStoreLimit(newMemstoreSize);
          for (HeapMemoryTuneObserver observer : tuneObservers) {
            // Risky.. If this newMemstoreSize decreases we reduce the count in offheap chunk pool
            observer.onHeapMemoryTune(newMemstoreSize, newBlockCacheSize);
          }
        }
      } else {
        metricsHeapMemoryManager.increaseTunerDoNothingCounter();
        if (LOG.isDebugEnabled()) {
          LOG.debug("No changes made by HeapMemoryTuner.");
        }
      }
    }

    @Override
    public void flushRequested(FlushType type, Region region) {
      switch (type) {
        case ABOVE_ONHEAP_HIGHER_MARK:
          blockedFlushCount.incrementAndGet();
          break;
        case ABOVE_ONHEAP_LOWER_MARK:
          unblockedFlushCount.incrementAndGet();
          break;
        // Removed the counting of the offheap related flushes (after reviews). Will add later if
        // needed
        default:
          // In case of any other flush don't do any action.
          break;
      }
    }
  }

  /**
   * POJO to pass all the relevant information required to do the heap memory tuning. It holds the
   * flush counts and block cache evictions happened within the interval. Also holds the current
   * heap percentage allocated for memstore and block cache.
   */
  public static final class TunerContext {
    private long blockedFlushCount;
    private long unblockedFlushCount;
    private long evictCount;
    private long cacheMissCount;
    private float curBlockCacheUsed;
    private float curMemStoreUsed;
    private float curMemStoreSize;
    private float curBlockCacheSize;
    private boolean offheapMemstore;

    public long getBlockedFlushCount() {
      return blockedFlushCount;
    }

    public void setBlockedFlushCount(long blockedFlushCount) {
      this.blockedFlushCount = blockedFlushCount;
    }

    public long getUnblockedFlushCount() {
      return unblockedFlushCount;
    }

    public void setUnblockedFlushCount(long unblockedFlushCount) {
      this.unblockedFlushCount = unblockedFlushCount;
    }

    public long getEvictCount() {
      return evictCount;
    }

    public void setEvictCount(long evictCount) {
      this.evictCount = evictCount;
    }

    public float getCurMemStoreSize() {
      return curMemStoreSize;
    }

    public void setCurMemStoreSize(float curMemStoreSize) {
      this.curMemStoreSize = curMemStoreSize;
    }

    public float getCurBlockCacheSize() {
      return curBlockCacheSize;
    }

    public void setCurBlockCacheSize(float curBlockCacheSize) {
      this.curBlockCacheSize = curBlockCacheSize;
    }

    public long getCacheMissCount() {
      return cacheMissCount;
    }

    public void setCacheMissCount(long cacheMissCount) {
      this.cacheMissCount = cacheMissCount;
    }

    public float getCurBlockCacheUsed() {
      return curBlockCacheUsed;
    }

    public void setCurBlockCacheUsed(float curBlockCacheUsed) {
      this.curBlockCacheUsed = curBlockCacheUsed;
    }

    public float getCurMemStoreUsed() {
      return curMemStoreUsed;
    }

    public void setCurMemStoreUsed(float d) {
      this.curMemStoreUsed = d;
    }

    public void setOffheapMemStore(boolean offheapMemstore) {
      this.offheapMemstore = offheapMemstore;
    }

    public boolean isOffheapMemStore() {
      return this.offheapMemstore;
    }
  }

  /**
   * POJO which holds the result of memory tuning done by HeapMemoryTuner implementation. It
   * includes the new heap percentage for memstore and block cache.
   */
  public static final class TunerResult {
    private float memstoreSize;
    private float blockCacheSize;
    private final boolean needsTuning;

    public TunerResult(boolean needsTuning) {
      this.needsTuning = needsTuning;
    }

    public float getMemStoreSize() {
      return memstoreSize;
    }

    public void setMemStoreSize(float memstoreSize) {
      this.memstoreSize = memstoreSize;
    }

    public float getBlockCacheSize() {
      return blockCacheSize;
    }

    public void setBlockCacheSize(float blockCacheSize) {
      this.blockCacheSize = blockCacheSize;
    }

    public boolean needsTuning() {
      return needsTuning;
    }
  }

  /**
   * Every class that wants to observe heap memory tune actions must implement this interface.
   */
  public static interface HeapMemoryTuneObserver {

    /**
     * This method would be called by HeapMemoryManger when a heap memory tune action took place.
     * @param newMemstoreSize   The newly calculated global memstore size
     * @param newBlockCacheSize The newly calculated global blockcache size
     */
    void onHeapMemoryTune(long newMemstoreSize, long newBlockCacheSize);
  }
}