diff --git a/ambari-metrics-timelineservice/pom.xml b/ambari-metrics-timelineservice/pom.xml
index c00ff69..ae4ecd9 100644
--- a/ambari-metrics-timelineservice/pom.xml
+++ b/ambari-metrics-timelineservice/pom.xml
@@ -327,6 +327,12 @@
       <groupId>org.apache.zookeeper</groupId>
       <artifactId>zookeeper</artifactId>
       <version>3.4.5.1.3.0.0-107</version>
+      <exclusions>
+        <exclusion>
+          <artifactId>jline</artifactId>
+          <groupId>jline</groupId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.phoenix</groupId>
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/HBaseTimelineMetricsService.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/HBaseTimelineMetricsService.java
index 56a28dc..d09d4bb 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/HBaseTimelineMetricsService.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/HBaseTimelineMetricsService.java
@@ -17,7 +17,6 @@
  */
 package org.apache.ambari.metrics.core.timeline;
 
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.DEFAULT_TOPN_HOSTS_LIMIT;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.USE_GROUPBY_AGGREGATOR_QUERIES;
 import static org.apache.ambari.metrics.core.timeline.availability.AggregationTaskRunner.ACTUAL_AGGREGATOR_NAMES;
 
@@ -35,7 +34,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -111,15 +109,18 @@
   private synchronized void initializeSubsystem() {
     if (!isInitialized) {
       hBaseAccessor = new PhoenixHBaseAccessor(null);
-      // Initialize schema
-      hBaseAccessor.initMetricSchema();
-      // Initialize metadata from store
+
+      // Initialize metadata
       try {
         metricMetadataManager = new TimelineMetricMetadataManager(hBaseAccessor);
       } catch (MalformedURLException | URISyntaxException e) {
         throw new ExceptionInInitializerError("Unable to initialize metadata manager");
       }
       metricMetadataManager.initializeMetadata();
+
+      // Initialize metric schema
+      hBaseAccessor.initMetricSchema();
+
       // Initialize policies before TTL update
       hBaseAccessor.initPoliciesAndTTL();
       // Start HA service
@@ -395,6 +396,10 @@
     return metricsFunctions;
   }
 
+  public void putMetricsSkipCache(TimelineMetrics metrics) throws SQLException, IOException {
+    hBaseAccessor.insertMetricRecordsWithMetadata(metricMetadataManager, metrics, true);
+  }
+
   @Override
   public TimelinePutResponse putMetrics(TimelineMetrics metrics) throws SQLException, IOException {
     // Error indicated by the Sql exception
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessor.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessor.java
index 040df1b..c684b0a 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessor.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessor.java
@@ -20,6 +20,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.ambari.metrics.core.timeline.FunctionUtils.findMetricFunctions;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.AGGREGATORS_SKIP_BLOCK_CACHE;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.BLOCKING_STORE_FILES_KEY;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_MINUTE_SLEEP_INTERVAL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CLUSTER_DAILY_TABLE_TTL;
@@ -27,41 +28,32 @@
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CLUSTER_MINUTE_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CLUSTER_SECOND_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.CONTAINER_METRICS_TTL;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.DATE_TIERED_COMPACTION_POLICY;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.FIFO_COMPACTION_POLICY_CLASS;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.GLOBAL_MAX_RETRIES;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.GLOBAL_RESULT_LIMIT;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.GLOBAL_RETRY_INTERVAL;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_BLOCKING_STORE_FILES;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_COMPRESSION_SCHEME;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_ENCODING_SCHEME;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_MINUTE_SLEEP_INTERVAL;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HSTORE_COMPACTION_CLASS_KEY;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HSTORE_ENGINE_CLASS;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TRANSIENT_METRIC_PATTERNS;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HOST_DAILY_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HOST_HOUR_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HOST_MINUTE_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.METRICS_TRANSIENT_TABLE_TTL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.PRECISION_TABLE_TTL;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_AGGREGATE_TABLES_DURABILITY;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_AGGREGATE_TABLE_HBASE_BLOCKING_STORE_FILES;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CACHE_COMMIT_INTERVAL;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CACHE_ENABLED;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CACHE_SIZE;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_CLASS;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_KEY;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_CLASS;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_KEY;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_PRECISION_TABLE_DURABILITY;
-import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_PRECISION_TABLE_HBASE_BLOCKING_STORE_FILES;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_AGGREGATOR_SINK_CLASS;
-import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.ALTER_METRICS_METADATA_TABLE;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CONTAINER_METRICS_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_CONTAINER_METRICS_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_TRANSIENT_METRICS_TABLE_SQL;
-import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_HOSTED_APPS_METADATA_TABLE_SQL;
-import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_INSTANCE_HOST_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_AGGREGATE_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_CLUSTER_AGGREGATE_GROUPED_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_CLUSTER_AGGREGATE_TABLE_SQL;
-import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_METADATA_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_TABLE_SQL;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_ENCODING;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_TABLE_COMPRESSION;
@@ -132,6 +124,7 @@
 import org.apache.ambari.metrics.core.timeline.source.InternalMetricsSource;
 import org.apache.ambari.metrics.core.timeline.source.InternalSourceProvider;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -201,21 +194,8 @@
   private TimelineMetricsAggregatorSink aggregatorSink;
   private final int cacheCommitInterval;
   private final boolean skipBlockCacheForAggregatorsEnabled;
-  private final String timelineMetricsTablesDurability;
-  private final String timelineMetricsPrecisionTableDurability;
   private TimelineMetricMetadataManager metadataManagerInstance;
 
-  static final String HSTORE_COMPACTION_CLASS_KEY =
-    "hbase.hstore.defaultengine.compactionpolicy.class";
-  static final String HSTORE_ENGINE_CLASS =
-    "hbase.hstore.engine.class";
-  static final String FIFO_COMPACTION_POLICY_CLASS =
-    "org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy";
-  static final String DATE_TIERED_COMPACTION_POLICY =
-    "org.apache.hadoop.hbase.regionserver.DateTieredStoreEngine";
-  static final String BLOCKING_STORE_FILES_KEY =
-    "hbase.hstore.blockingStoreFiles";
-
   private Map<String, Integer> tableTTL = new HashMap<>();
 
   private final TimelineMetricConfiguration configuration;
@@ -259,11 +239,9 @@
     this.cacheCommitInterval = Integer.valueOf(metricsConf.get(TIMELINE_METRICS_CACHE_COMMIT_INTERVAL, "3"));
     this.insertCache = new ArrayBlockingQueue<TimelineMetrics>(cacheSize);
     this.skipBlockCacheForAggregatorsEnabled = metricsConf.getBoolean(AGGREGATORS_SKIP_BLOCK_CACHE, false);
-    this.timelineMetricsTablesDurability = metricsConf.get(TIMELINE_METRICS_AGGREGATE_TABLES_DURABILITY, "");
-    this.timelineMetricsPrecisionTableDurability = metricsConf.get(TIMELINE_METRICS_PRECISION_TABLE_DURABILITY, "");
 
     tableTTL.put(METRICS_RECORD_TABLE_NAME, metricsConf.getInt(PRECISION_TABLE_TTL, 1 * 86400));  // 1 day
-    tableTTL.put(CONTAINER_METRICS_TABLE_NAME, metricsConf.getInt(CONTAINER_METRICS_TTL, 30 * 86400));  // 30 days
+    tableTTL.put(CONTAINER_METRICS_TABLE_NAME, metricsConf.getInt(CONTAINER_METRICS_TTL, 14 * 86400));  // 30 days
     tableTTL.put(METRICS_AGGREGATE_MINUTE_TABLE_NAME, metricsConf.getInt(HOST_MINUTE_TABLE_TTL, 7 * 86400)); //7 days
     tableTTL.put(METRICS_AGGREGATE_HOURLY_TABLE_NAME, metricsConf.getInt(HOST_HOUR_TABLE_TTL, 30 * 86400)); //30 days
     tableTTL.put(METRICS_AGGREGATE_DAILY_TABLE_NAME, metricsConf.getInt(HOST_DAILY_TABLE_TTL, 365 * 86400)); //1 year
@@ -470,7 +448,7 @@
     return mapper.readValue(json, metricValuesTypeRef);
   }
 
-  private Connection getConnectionRetryingOnException() throws SQLException, InterruptedException {
+  public Connection getConnectionRetryingOnException() throws SQLException, InterruptedException {
     RetryCounter retryCounter = retryCounterFactory.create();
     while (true) {
       try{
@@ -511,6 +489,10 @@
   protected void initMetricSchema() {
     Connection conn = null;
     Statement stmt = null;
+    PreparedStatement pStmt = null;
+    TimelineMetricSplitPointComputer splitPointComputer = new TimelineMetricSplitPointComputer(
+      metricsConf, hbaseConf, metadataManagerInstance);
+    splitPointComputer.computeSplitPoints();
 
     String encoding = metricsConf.get(HBASE_ENCODING_SCHEME, DEFAULT_ENCODING);
     String compression = metricsConf.get(HBASE_COMPRESSION_SCHEME, DEFAULT_TABLE_COMPRESSION);
@@ -521,21 +503,6 @@
       conn = getConnectionRetryingOnException();
       stmt = conn.createStatement();
 
-      // Metadata
-      String metadataSql = String.format(CREATE_METRICS_METADATA_TABLE_SQL,
-        encoding, compression);
-      stmt.executeUpdate(metadataSql);
-      stmt.executeUpdate(ALTER_METRICS_METADATA_TABLE);
-
-      String hostedAppSql = String.format(CREATE_HOSTED_APPS_METADATA_TABLE_SQL,
-        encoding, compression);
-      stmt.executeUpdate(hostedAppSql);
-
-      //Host Instances table
-      String hostedInstancesSql = String.format(CREATE_INSTANCE_HOST_TABLE_SQL,
-        encoding, compression);
-      stmt.executeUpdate(hostedInstancesSql);
-
       // Container Metrics
       stmt.executeUpdate( String.format(CREATE_CONTAINER_METRICS_TABLE_SQL,
         encoding, tableTTL.get(CONTAINER_METRICS_TABLE_NAME), compression));
@@ -543,13 +510,15 @@
       // Host level
       String precisionSql = String.format(CREATE_METRICS_TABLE_SQL,
         encoding, tableTTL.get(METRICS_RECORD_TABLE_NAME), compression);
-      stmt.executeUpdate(precisionSql);
+      pStmt = prepareCreateMetricsTableStatement(conn, precisionSql, splitPointComputer.getPrecisionSplitPoints());
+      pStmt.executeUpdate();
 
       String hostMinuteAggregrateSql = String.format(CREATE_METRICS_AGGREGATE_TABLE_SQL,
         METRICS_AGGREGATE_MINUTE_TABLE_NAME, encoding,
         tableTTL.get(METRICS_AGGREGATE_MINUTE_TABLE_NAME),
         compression);
-      stmt.executeUpdate(hostMinuteAggregrateSql);
+      pStmt = prepareCreateMetricsTableStatement(conn, hostMinuteAggregrateSql, splitPointComputer.getHostAggregateSplitPoints());
+      pStmt.executeUpdate();
 
       stmt.executeUpdate(String.format(CREATE_METRICS_AGGREGATE_TABLE_SQL,
         METRICS_AGGREGATE_HOURLY_TABLE_NAME, encoding,
@@ -565,8 +534,9 @@
         METRICS_CLUSTER_AGGREGATE_TABLE_NAME, encoding,
         tableTTL.get(METRICS_CLUSTER_AGGREGATE_TABLE_NAME),
         compression);
+      pStmt = prepareCreateMetricsTableStatement(conn, aggregateSql, splitPointComputer.getClusterAggregateSplitPoints());
+      pStmt.executeUpdate();
 
-      stmt.executeUpdate(aggregateSql);
       stmt.executeUpdate(String.format(CREATE_METRICS_CLUSTER_AGGREGATE_GROUPED_TABLE_SQL,
         METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME, encoding,
         tableTTL.get(METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME),
@@ -603,6 +573,13 @@
           // Ignore
         }
       }
+      if (pStmt != null) {
+        try {
+          pStmt.close();
+        } catch (Exception e) {
+          // Ignore
+        }
+      }
       if (conn != null) {
         try {
           conn.close();
@@ -613,7 +590,7 @@
     }
   }
 
-  protected void initPoliciesAndTTL() {
+  void initPoliciesAndTTL() {
     Admin hBaseAdmin = null;
     try {
       hBaseAdmin = dataSource.getHBaseAdmin();
@@ -622,9 +599,13 @@
     }
 
     TableName[] tableNames = null;
+    TableName[] containerMetricsTableName = null;
+
     if (hBaseAdmin != null) {
       try {
         tableNames = hBaseAdmin.listTableNames(PHOENIX_TABLES_REGEX_PATTERN, false);
+        containerMetricsTableName = hBaseAdmin.listTableNames(CONTAINER_METRICS_TABLE_NAME, false);
+        tableNames = (TableName[]) ArrayUtils.addAll(tableNames, containerMetricsTableName);
       } catch (IOException e) {
         LOG.warn("Unable to get table names from HBaseAdmin for setting policies.", e);
         return;
@@ -708,72 +689,44 @@
   }
 
   private boolean setDurabilityForTable(String tableName, TableDescriptorBuilder tableDescriptor) {
-
-    boolean modifyTable = false;
-
-    if (METRIC_TRANSIENT_TABLE_NAME.equalsIgnoreCase(tableName)) {
-      tableDescriptor.setDurability(Durability.SKIP_WAL);
-      modifyTable = true;
-    } else if (METRICS_RECORD_TABLE_NAME.equals(tableName)) {
-      if (!timelineMetricsPrecisionTableDurability.isEmpty()) {
-        LOG.info("Setting WAL option " + timelineMetricsPrecisionTableDurability + " for table : " + tableName);
-        boolean validDurability = true;
-        if ("SKIP_WAL".equals(timelineMetricsPrecisionTableDurability)) {
-          tableDescriptor.setDurability(Durability.SKIP_WAL);
-        } else if ("SYNC_WAL".equals(timelineMetricsPrecisionTableDurability)) {
-          tableDescriptor.setDurability(Durability.SYNC_WAL);
-        } else if ("ASYNC_WAL".equals(timelineMetricsPrecisionTableDurability)) {
-          tableDescriptor.setDurability(Durability.ASYNC_WAL);
-        } else if ("FSYNC_WAL".equals(timelineMetricsPrecisionTableDurability)) {
-          tableDescriptor.setDurability(Durability.FSYNC_WAL);
-        } else {
-          LOG.info("Unknown value for " + TIMELINE_METRICS_PRECISION_TABLE_DURABILITY + " : " + timelineMetricsPrecisionTableDurability);
-          validDurability = false;
-        }
-        if (validDurability) {
-          modifyTable = true;
-        }
+    String tableDurability = metricsConf.get("timeline.metrics." + tableName + ".durability", "");
+    if (StringUtils.isNotEmpty(tableDurability)) {
+      LOG.info("Setting WAL option " + tableDurability + " for table : " + tableName);
+      boolean validDurability = true;
+      if ("SKIP_WAL".equals(tableDurability)) {
+        tableDescriptor.setDurability(Durability.SKIP_WAL);
+      } else if ("SYNC_WAL".equals(tableDurability)) {
+        tableDescriptor.setDurability(Durability.SYNC_WAL);
+      } else if ("ASYNC_WAL".equals(tableDurability)) {
+        tableDescriptor.setDurability(Durability.ASYNC_WAL);
+      } else if ("FSYNC_WAL".equals(tableDurability)) {
+        tableDescriptor.setDurability(Durability.FSYNC_WAL);
+      } else {
+        LOG.info("Unknown value for durability : " + tableDurability);
+        validDurability = false;
       }
-    } else {
-      if (!timelineMetricsTablesDurability.isEmpty()) {
-        LOG.info("Setting WAL option " + timelineMetricsTablesDurability + " for table : " + tableName);
-        boolean validDurability = true;
-        if ("SKIP_WAL".equals(timelineMetricsTablesDurability)) {
-          tableDescriptor.setDurability(Durability.SKIP_WAL);
-        } else if ("SYNC_WAL".equals(timelineMetricsTablesDurability)) {
-          tableDescriptor.setDurability(Durability.SYNC_WAL);
-        } else if ("ASYNC_WAL".equals(timelineMetricsTablesDurability)) {
-          tableDescriptor.setDurability(Durability.ASYNC_WAL);
-        } else if ("FSYNC_WAL".equals(timelineMetricsTablesDurability)) {
-          tableDescriptor.setDurability(Durability.FSYNC_WAL);
-        } else {
-          LOG.info("Unknown value for " + TIMELINE_METRICS_AGGREGATE_TABLES_DURABILITY + " : " + timelineMetricsTablesDurability);
-          validDurability = false;
-        }
-        if (validDurability) {
-          modifyTable = true;
-        }
-      }
+      return validDurability;
     }
-    return modifyTable;
+    return false;
   }
 
+
   private boolean setCompactionPolicyForTable(String tableName, TableDescriptorBuilder tableDescriptorBuilder) {
 
     boolean modifyTable = false;
 
-    String compactionPolicyKey = metricsConf.get(TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_KEY,
-      HSTORE_ENGINE_CLASS);
-    String compactionPolicyClass = metricsConf.get(TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_CLASS,
-      DATE_TIERED_COMPACTION_POLICY);
-    int blockingStoreFiles = hbaseConf.getInt(TIMELINE_METRICS_AGGREGATE_TABLE_HBASE_BLOCKING_STORE_FILES, 60);
+    String keyConfig = "timeline.metrics." + tableName + ".compaction.policy.key";
+    String policyConfig = "timeline.metrics." + tableName + ".compaction.policy";
+    String storeFilesConfig = "timeline.metrics." + tableName + ".blocking.store.files";
 
-    if (tableName.equals(METRICS_RECORD_TABLE_NAME) || tableName.equalsIgnoreCase(METRIC_TRANSIENT_TABLE_NAME)) {
-      compactionPolicyKey = metricsConf.get(TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_KEY,
-        HSTORE_COMPACTION_CLASS_KEY);
-      compactionPolicyClass = metricsConf.get(TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_CLASS,
-        FIFO_COMPACTION_POLICY_CLASS);
-      blockingStoreFiles = hbaseConf.getInt(TIMELINE_METRICS_PRECISION_TABLE_HBASE_BLOCKING_STORE_FILES, 1000);
+    String compactionPolicyKey = metricsConf.get(keyConfig, HSTORE_ENGINE_CLASS);
+    String compactionPolicyClass = metricsConf.get(policyConfig, DATE_TIERED_COMPACTION_POLICY);
+    int blockingStoreFiles = hbaseConf.getInt(storeFilesConfig, 60);
+
+    if (tableName.equals(METRICS_RECORD_TABLE_NAME)) {
+      compactionPolicyKey = metricsConf.get(keyConfig, HSTORE_COMPACTION_CLASS_KEY);
+      compactionPolicyClass = metricsConf.get(policyConfig, FIFO_COMPACTION_POLICY_CLASS);
+      blockingStoreFiles = hbaseConf.getInt(storeFilesConfig, 1000);
     }
 
     if (StringUtils.isEmpty(compactionPolicyKey) || StringUtils.isEmpty(compactionPolicyClass)) {
@@ -781,46 +734,54 @@
       modifyTable = setHbaseBlockingStoreFiles(tableDescriptorBuilder, tableName, 300);
     } else {
       tableDescriptorBuilder.setValue(compactionPolicyKey, compactionPolicyClass);
-      tableDescriptorBuilder.removeValue(HSTORE_ENGINE_CLASS.getBytes());
-      tableDescriptorBuilder.removeValue(HSTORE_COMPACTION_CLASS_KEY.getBytes());
       setHbaseBlockingStoreFiles(tableDescriptorBuilder, tableName, blockingStoreFiles);
       modifyTable = true;
     }
 
+    if (!compactionPolicyKey.equals(HSTORE_ENGINE_CLASS)) {
+      tableDescriptorBuilder.removeValue(HSTORE_ENGINE_CLASS.getBytes());
+    }
+    if (!compactionPolicyKey.equals(HSTORE_COMPACTION_CLASS_KEY)) {
+      tableDescriptorBuilder.removeValue(HSTORE_COMPACTION_CLASS_KEY.getBytes());
+    }
+
     return modifyTable;
   }
 
   private boolean setHbaseBlockingStoreFiles(TableDescriptorBuilder tableDescriptor, String tableName, int value) {
-    int blockingStoreFiles = hbaseConf.getInt(HBASE_BLOCKING_STORE_FILES, value);
-    if (blockingStoreFiles != value) {
-      blockingStoreFiles = value;
-      tableDescriptor.setValue(BLOCKING_STORE_FILES_KEY, String.valueOf(value));
-      LOG.info("Setting config property " + BLOCKING_STORE_FILES_KEY +
-        " = " + blockingStoreFiles + " for " + tableName);
-      return true;
-    }
-    return false;
+    tableDescriptor.setValue(BLOCKING_STORE_FILES_KEY, String.valueOf(value));
+    LOG.info("Setting config property " + BLOCKING_STORE_FILES_KEY +
+      " = " + value + " for " + tableName);
+    return true;
   }
 
-  protected String getSplitPointsStr(String splitPoints) {
-    if (StringUtils.isEmpty(splitPoints.trim())) {
+
+  private PreparedStatement prepareCreateMetricsTableStatement(Connection connection,
+                                                               String sql,
+                                                               List<byte[]> splitPoints) throws SQLException {
+
+    String createTableWithSplitPointsSql = sql + getSplitPointsStr(splitPoints.size());
+    LOG.info(createTableWithSplitPointsSql);
+    PreparedStatement statement = connection.prepareStatement(createTableWithSplitPointsSql);
+    for (int i = 1; i <= splitPoints.size(); i++) {
+      statement.setBytes(i, splitPoints.get(i - 1));
+    }
+    return statement;
+  }
+
+  private String getSplitPointsStr(int numSplits) {
+    if (numSplits <= 0) {
       return "";
     }
-    String[] points = splitPoints.split(",");
-    if (points.length > 0) {
-      StringBuilder sb = new StringBuilder(" SPLIT ON ");
-      sb.append("(");
-      for (String point : points) {
-        sb.append("'");
-        sb.append(point.trim());
-        sb.append("'");
-        sb.append(",");
-      }
-      sb.deleteCharAt(sb.length() - 1);
-      sb.append(")");
-      return sb.toString();
+    StringBuilder sb = new StringBuilder(" SPLIT ON ");
+    sb.append("(");
+    for (int i = 0; i < numSplits; i++) {
+      sb.append("?");
+      sb.append(",");
     }
-    return "";
+    sb.deleteCharAt(sb.length() - 1);
+    sb.append(")");
+    return sb.toString();
   }
 
   /**
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricConfiguration.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricConfiguration.java
index 6ec2c6b..393d4a3 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricConfiguration.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricConfiguration.java
@@ -237,12 +237,6 @@
   public static final String WATCHER_MAX_FAILURES =
     "timeline.metrics.service.watcher.max.failures";
 
-  public static final String PRECISION_TABLE_SPLIT_POINTS =
-    "timeline.metrics.host.aggregate.splitpoints";
-
-  public static final String AGGREGATE_TABLE_SPLIT_POINTS =
-    "timeline.metrics.cluster.aggregate.splitpoints";
-
   public static final String AGGREGATORS_SKIP_BLOCK_CACHE =
     "timeline.metrics.aggregators.skip.blockcache.enabled";
 
@@ -261,12 +255,6 @@
   public static final String TIMELINE_METRICS_SINK_COLLECTION_PERIOD =
     "timeline.metrics.sink.collection.period";
 
-  public static final String TIMELINE_METRICS_PRECISION_TABLE_DURABILITY =
-    "timeline.metrics.precision.table.durability";
-
-  public static final String TIMELINE_METRICS_AGGREGATE_TABLES_DURABILITY =
-      "timeline.metrics.aggregate.tables.durability";
-
   public static final String TIMELINE_METRICS_WHITELIST_ENABLED =
     "timeline.metrics.whitelisting.enabled";
 
@@ -285,33 +273,9 @@
   public static final String TIMELINE_METRICS_APPS_WHITELIST =
     "timeline.metrics.apps.whitelist";
 
-  public static final String HBASE_BLOCKING_STORE_FILES =
-    "hbase.hstore.blockingStoreFiles";
-
-  public static final String DEFAULT_TOPN_HOSTS_LIMIT =
-    "timeline.metrics.default.topn.hosts.limit";
-
   public static final String TIMELINE_METRIC_AGGREGATION_SQL_FILTERS =
     "timeline.metrics.cluster.aggregation.sql.filters";
 
-  public static final String TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_KEY =
-    "timeline.metrics.hbase.aggregate.table.compaction.policy.key";
-
-  public static final String TIMELINE_METRICS_HBASE_AGGREGATE_TABLE_COMPACTION_POLICY_CLASS =
-    "timeline.metrics.hbase.aggregate.table.compaction.policy.class";
-
-  public static final String TIMELINE_METRICS_AGGREGATE_TABLE_HBASE_BLOCKING_STORE_FILES =
-    "timeline.metrics.aggregate.table.hbase.hstore.blockingStoreFiles";
-
-  public static final String TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_KEY =
-    "timeline.metrics.hbase.precision.table.compaction.policy.key";
-
-  public static final String TIMELINE_METRICS_HBASE_PRECISION_TABLE_COMPACTION_POLICY_CLASS =
-    "timeline.metrics.hbase.precision.table.compaction.policy.class";
-
-  public static final String TIMELINE_METRICS_PRECISION_TABLE_HBASE_BLOCKING_STORE_FILES =
-    "timeline.metrics.precision.table.hbase.hstore.blockingStoreFiles";
-
   public static final String TIMELINE_METRICS_SUPPORT_MULTIPLE_CLUSTERS =
     "timeline.metrics.support.multiple.clusters";
 
@@ -346,6 +310,9 @@
 
   public static final String TRANSIENT_METRIC_PATTERNS = "timeline.metrics.transient.metric.patterns";
 
+  public static final String TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS = "timeline.metrics.initial.configured.master.components";
+  public static final String TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS = "timeline.metrics.initial.configured.slave.components";
+
   public static final String KAFKA_SERVERS = "timeline.metrics.external.sink.kafka.bootstrap.servers";
   public static final String KAFKA_ACKS = "timeline.metrics.external.sink.kafka.acks";
   public static final String KAFKA_RETRIES = "timeline.metrics.external.sink.kafka.bootstrap.retries";
@@ -353,7 +320,13 @@
   public static final String KAFKA_LINGER_MS = "timeline.metrics.external.sink.kafka.linger.ms";
   public static final String KAFKA_BUFFER_MEM = "timeline.metrics.external.sink.kafka.buffer.memory";
   public static final String KAFKA_SINK_TIMEOUT_SECONDS = "timeline.metrics.external.sink.kafka.timeout.seconds";
-  
+
+  public static final String HSTORE_COMPACTION_CLASS_KEY = "hbase.hstore.defaultengine.compactionpolicy.class";
+  public static final String HSTORE_ENGINE_CLASS = "hbase.hstore.engine.class";
+  public static final String FIFO_COMPACTION_POLICY_CLASS = "org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy";
+  public static final String DATE_TIERED_COMPACTION_POLICY = "org.apache.hadoop.hbase.regionserver.DateTieredStoreEngine";
+  public static final String BLOCKING_STORE_FILES_KEY = "hbase.hstore.blockingStoreFiles";
+
   private Configuration hbaseConf;
   private Configuration metricsConf;
   private Configuration metricsSslConf;
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputer.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputer.java
new file mode 100644
index 0000000..05366cc
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputer.java
@@ -0,0 +1,240 @@
+/**
+ * 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.ambari.metrics.core.timeline;
+
+import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
+import org.apache.ambari.metrics.core.timeline.discovery.TimelineMetricMetadataManager;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS;
+
+public class TimelineMetricSplitPointComputer {
+
+  private static final Log LOG = LogFactory.getLog(TimelineMetricSplitPointComputer.class);
+  private Set<String> masterComponents = new HashSet<>();
+  private Set<String> slaveComponents = new HashSet<>();
+
+  private static final int MINIMUM_PRECISION_TABLE_REGIONS = 4;
+  private static final int MINIMUM_AGGREGATE_TABLE_REGIONS = 2;
+  private static final int OTHER_TABLE_STATIC_REGIONS = 8;
+  private static final int SLAVE_EQUIDISTANT_POINTS = 50;
+  private static final int MASTER_EQUIDISTANT_POINTS = 5;
+
+  private double hbaseTotalHeapsize;
+  private double hbaseMemstoreUpperLimit;
+  private double hbaseMemstoreFlushSize;
+  private TimelineMetricMetadataManager timelineMetricMetadataManager = null;
+
+  private List<byte[]> precisionSplitPoints = new ArrayList<>();
+  private List<byte[]> aggregateSplitPoints = new ArrayList<>();
+
+  public TimelineMetricSplitPointComputer(Configuration metricsConf,
+                                          Configuration hbaseConf,
+                                          TimelineMetricMetadataManager timelineMetricMetadataManager) {
+
+    String componentsString = metricsConf.get(TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS, "");
+    if (StringUtils.isNotEmpty(componentsString)) {
+      masterComponents.addAll(Arrays.asList(componentsString.split(",")));
+    }
+
+   componentsString = metricsConf.get(TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS, "");
+    if (StringUtils.isNotEmpty(componentsString)) {
+      slaveComponents.addAll(Arrays.asList(componentsString.split(",")));
+    }
+
+    this.timelineMetricMetadataManager = timelineMetricMetadataManager;
+    hbaseTotalHeapsize = metricsConf.getDouble("hbase_total_heapsize", 1024*1024*1024);
+    hbaseMemstoreUpperLimit = hbaseConf.getDouble("hbase.regionserver.global.memstore.upperLimit", 0.3);
+    hbaseMemstoreFlushSize = hbaseConf.getDouble("hbase.hregion.memstore.flush.size", 134217728);
+  }
+
+
+  protected void computeSplitPoints() {
+
+    double memstoreMaxMemory = hbaseMemstoreUpperLimit * hbaseTotalHeapsize;
+    int maxInMemoryRegions = (int) ((memstoreMaxMemory / hbaseMemstoreFlushSize) - OTHER_TABLE_STATIC_REGIONS);
+
+    int targetPrecisionTableRegionCount = MINIMUM_PRECISION_TABLE_REGIONS;
+    int targetAggregateTableRegionCount = MINIMUM_AGGREGATE_TABLE_REGIONS;
+
+    if (maxInMemoryRegions > 2) {
+      targetPrecisionTableRegionCount =  Math.max(4, (int)(0.70 * maxInMemoryRegions));
+      targetAggregateTableRegionCount =  Math.max(2, (int)(0.15 * maxInMemoryRegions));
+    }
+
+    List<MetricApp> metricList = new ArrayList<>();
+
+    for (String component : masterComponents) {
+      metricList.addAll(getSortedMetricListForSplitPoint(component, false));
+    }
+
+    for (String component : slaveComponents) {
+      metricList.addAll(getSortedMetricListForSplitPoint(component, true));
+    }
+
+    int totalMetricLength = metricList.size();
+
+    if (targetPrecisionTableRegionCount > 1) {
+      int idx = (int) Math.ceil(totalMetricLength / targetPrecisionTableRegionCount);
+      int index = idx;
+      for (int i = 0; i < targetPrecisionTableRegionCount; i++) {
+        if (index < totalMetricLength - 1) {
+          MetricApp metricAppService = metricList.get(index);
+          byte[] uuid = timelineMetricMetadataManager.getUuid(
+            new TimelineClusterMetric(metricAppService.metricName, metricAppService.appId, null, -1),
+            true);
+          precisionSplitPoints.add(uuid);
+          index += idx;
+        }
+      }
+    }
+
+    if (targetAggregateTableRegionCount > 1) {
+      int idx = (int) Math.ceil(totalMetricLength / targetAggregateTableRegionCount);
+      int index = idx;
+      for (int i = 0; i < targetAggregateTableRegionCount; i++) {
+        if (index < totalMetricLength - 1) {
+          MetricApp metricAppService = metricList.get(index);
+          byte[] uuid = timelineMetricMetadataManager.getUuid(
+            new TimelineClusterMetric(metricAppService.metricName, metricAppService.appId, null, -1),
+            true);
+          aggregateSplitPoints.add(uuid);
+          index += idx;
+        }
+      }
+    }
+  }
+
+  private List<MetricApp> getSortedMetricListForSplitPoint(String component, boolean isSlave) {
+
+    String appId = getAppId(component);
+    List<MetricApp> metricList = new ArrayList<>();
+
+    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+    if (classLoader == null) {
+      classLoader = getClass().getClassLoader();
+    }
+
+    String strLine;
+    BufferedReader bufferedReader;
+
+    try (InputStream inputStream = classLoader.getResourceAsStream("metrics_def/" + appId.toUpperCase() + ".dat")) {
+
+      if (inputStream != null) {
+        bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+        LOG.info("Found split point candidate metrics for : " + appId);
+
+        while ((strLine = bufferedReader.readLine()) != null) {
+          metricList.add(new MetricApp(strLine.trim(), appId));
+        }
+      } else {
+        LOG.info("Split point candidate metrics not found for : " + appId);
+      }
+    } catch (Exception e) {
+      LOG.info("Error reading split point candidate metrics for component : " + component);
+      LOG.error(e);
+    }
+
+    if (isSlave) {
+      return getEquidistantMetrics(metricList, SLAVE_EQUIDISTANT_POINTS);
+    } else {
+      return getEquidistantMetrics(metricList, MASTER_EQUIDISTANT_POINTS);
+    }
+  }
+
+  private List<MetricApp> getEquidistantMetrics(List<MetricApp> metrics, int distance) {
+    List<MetricApp> selectedMetricApps = new ArrayList<>();
+
+    int idx = metrics.size() / distance;
+    if (idx == 0) {
+      return metrics;
+    }
+
+    int index = idx;
+    for (int i = 0; i < distance; i++) {
+      selectedMetricApps.add(metrics.get(index - 1));
+      index += idx;
+    }
+    return selectedMetricApps;
+  }
+
+
+  public List<byte[]> getPrecisionSplitPoints() {
+    return precisionSplitPoints;
+  }
+
+  public List<byte[]> getClusterAggregateSplitPoints() {
+    return aggregateSplitPoints;
+  }
+
+  public List<byte[]> getHostAggregateSplitPoints() {
+    return aggregateSplitPoints;
+  }
+
+  private String getAppId(String component) {
+
+    if (component.equalsIgnoreCase("METRICS_COLLECTOR")) {
+      return "ams-hbase";
+    }
+
+    if (component.equalsIgnoreCase("METRICS_MONITOR")) {
+      return "HOST";
+    }
+    return component;
+  }
+}
+
+class MetricApp implements Comparable{
+  String metricName;
+  String appId;
+
+  MetricApp(String metricName, String appId) {
+    this.metricName = metricName;
+    if (appId.startsWith("hbase")) {
+      this.appId = "hbase";
+    } else {
+      this.appId = appId;
+    }
+  }
+
+  @Override
+  public int compareTo(Object o) {
+    MetricApp that = (MetricApp)o;
+
+    int metricCompare = metricName.compareTo(that.metricName);
+    if (metricCompare != 0) {
+      return metricCompare;
+    }
+
+    return appId.compareTo(that.appId);
+  }
+}
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcher.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcher.java
index 0ab7929..ba7ce44 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcher.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcher.java
@@ -45,13 +45,13 @@
   private static int failures = 0;
   private final TimelineMetricConfiguration configuration;
 
-  private TimelineMetricStore timelineMetricStore;
+  private HBaseTimelineMetricsService timelineMetricStore;
 
   //used to call timelineMetricStore blocking methods with timeout
   private ExecutorService executor = Executors.newSingleThreadExecutor();
 
 
-  public TimelineMetricStoreWatcher(TimelineMetricStore timelineMetricStore,
+  public TimelineMetricStoreWatcher(HBaseTimelineMetricsService timelineMetricStore,
                                     TimelineMetricConfiguration configuration) {
     this.timelineMetricStore = timelineMetricStore;
     this.configuration = configuration;
@@ -100,7 +100,7 @@
 
     Callable<TimelineMetric> task = new Callable<TimelineMetric>() {
       public TimelineMetric call() throws Exception {
-        timelineMetricStore.putMetrics(metrics);
+        timelineMetricStore.putMetricsSkipCache(metrics);
         TimelineMetrics timelineMetrics = timelineMetricStore.getTimelineMetrics(
           Collections.singletonList(FAKE_METRIC_NAME), Collections.singletonList(FAKE_HOSTNAME),
           FAKE_APP_ID, null, startTime - delay * 2 * 1000,
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/discovery/TimelineMetricMetadataManager.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/discovery/TimelineMetricMetadataManager.java
index 1ca5bc0..86226ec 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/discovery/TimelineMetricMetadataManager.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/discovery/TimelineMetricMetadataManager.java
@@ -20,7 +20,9 @@
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URISyntaxException;
+import java.sql.Connection;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -38,9 +40,12 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.apache.ambari.metrics.core.timeline.MetricsSystemInitializationException;
 import org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration;
 import org.apache.ambari.metrics.core.timeline.uuid.MetricUuidGenStrategy;
 import org.apache.ambari.metrics.core.timeline.uuid.MD5UuidGenStrategy;
+import org.apache.ambari.metrics.core.timeline.uuid.Murmur3HashUuidGenStrategy;
+import org.apache.ambari.metrics.core.timeline.uuid.TimelineMetricUuid;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.ArrayUtils;
@@ -53,13 +58,19 @@
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetricMetadata;
 import org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor;
 import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
-import org.apache.ambari.metrics.core.timeline.uuid.HashBasedUuidGenStrategy;
 
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_COMPRESSION_SCHEME;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HBASE_ENCODING_SCHEME;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TRANSIENT_METRIC_PATTERNS;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.METRICS_METADATA_SYNC_INIT_DELAY;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.METRICS_METADATA_SYNC_SCHEDULE_DELAY;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_UUID_GEN_STRATEGY;
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_METADATA_FILTERS;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_HOSTED_APPS_METADATA_TABLE_SQL;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_INSTANCE_HOST_TABLE_SQL;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CREATE_METRICS_METADATA_TABLE_SQL;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_ENCODING;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.DEFAULT_TABLE_COMPRESSION;
 import static org.apache.hadoop.metrics2.sink.timeline.TimelineMetricUtils.getJavaMetricPatterns;
 import static org.apache.hadoop.metrics2.sink.timeline.TimelineMetricUtils.getJavaRegexFromSqlRegex;
 
@@ -67,18 +78,18 @@
   private static final Log LOG = LogFactory.getLog(TimelineMetricMetadataManager.class);
   // Cache all metadata on retrieval
   private final Map<TimelineMetricMetadataKey, TimelineMetricMetadata> METADATA_CACHE = new ConcurrentHashMap<>();
-  private final Map<String, TimelineMetricMetadataKey> uuidKeyMap = new ConcurrentHashMap<>();
+  private final Map<TimelineMetricUuid, TimelineMetricMetadataKey> uuidKeyMap = new ConcurrentHashMap<>();
   // Map to lookup apps on a host
   private final Map<String, TimelineMetricHostMetadata> HOSTED_APPS_MAP = new ConcurrentHashMap<>();
-  private final Map<String, String> uuidHostMap = new ConcurrentHashMap<>();
+  private final Map<TimelineMetricUuid, String> uuidHostMap = new ConcurrentHashMap<>();
   private final Map<String, Set<String>> INSTANCE_HOST_MAP = new ConcurrentHashMap<>();
   // Sync only when needed
   AtomicBoolean SYNC_HOSTED_APPS_METADATA = new AtomicBoolean(false);
   AtomicBoolean SYNC_HOSTED_INSTANCES_METADATA = new AtomicBoolean(false);
 
-  private MetricUuidGenStrategy uuidGenStrategy = new HashBasedUuidGenStrategy();
+  private MetricUuidGenStrategy uuidGenStrategy = new Murmur3HashUuidGenStrategy();
   public static final int TIMELINE_METRIC_UUID_LENGTH = 16;
-  public static final int HOSTNAME_UUID_LENGTH = 16;
+  public static int HOSTNAME_UUID_LENGTH = 4;
 
   //Transient metric patterns. No UUID management and aggregation for such metrics.
   private List<String> transientMetricPatterns = new ArrayList<>();
@@ -120,7 +131,54 @@
    * Initialize Metadata from the store
    */
   public void initializeMetadata() {
-    metricMetadataSync = new TimelineMetricMetadataSync(this);
+
+    //Create metadata schema
+    Connection conn = null;
+    Statement stmt = null;
+
+    String encoding = metricsConf.get(HBASE_ENCODING_SCHEME, DEFAULT_ENCODING);
+    String compression = metricsConf.get(HBASE_COMPRESSION_SCHEME, DEFAULT_TABLE_COMPRESSION);
+
+    try {
+      LOG.info("Initializing metrics metadata schema...");
+      conn = hBaseAccessor.getConnectionRetryingOnException();
+      stmt = conn.createStatement();
+
+      // Metadata
+      String metadataSql = String.format(CREATE_METRICS_METADATA_TABLE_SQL,
+        encoding, compression);
+      stmt.executeUpdate(metadataSql);
+
+      String hostedAppSql = String.format(CREATE_HOSTED_APPS_METADATA_TABLE_SQL,
+        encoding, compression);
+      stmt.executeUpdate(hostedAppSql);
+
+      //Host Instances table
+      String hostedInstancesSql = String.format(CREATE_INSTANCE_HOST_TABLE_SQL,
+        encoding, compression);
+      stmt.executeUpdate(hostedInstancesSql);
+    } catch (SQLException | InterruptedException sql) {
+      LOG.error("Error creating Metrics Schema in HBase using Phoenix.", sql);
+      throw new MetricsSystemInitializationException(
+        "Error creating Metrics Metadata Schema in HBase using Phoenix.", sql);
+    } finally {
+      if (stmt != null) {
+        try {
+          stmt.close();
+        } catch (SQLException e) {
+          // Ignore
+        }
+      }
+      if (conn != null) {
+        try {
+          conn.close();
+        } catch (SQLException e) {
+          // Ignore
+        }
+      }
+    }
+
+      metricMetadataSync = new TimelineMetricMetadataSync(this);
     // Schedule the executor to sync to store
     executorService.scheduleWithFixedDelay(metricMetadataSync,
       metricsConf.getInt(METRICS_METADATA_SYNC_INIT_DELAY, 120), // 2 minutes
@@ -335,14 +393,14 @@
     for (TimelineMetricMetadataKey key : METADATA_CACHE.keySet()) {
       TimelineMetricMetadata timelineMetricMetadata = METADATA_CACHE.get(key);
       if (timelineMetricMetadata != null && timelineMetricMetadata.getUuid() != null) {
-        uuidKeyMap.put(new String(timelineMetricMetadata.getUuid()), key);
+        uuidKeyMap.put(new TimelineMetricUuid(timelineMetricMetadata.getUuid()), key);
       }
     }
 
     for (String host : HOSTED_APPS_MAP.keySet()) {
       TimelineMetricHostMetadata timelineMetricHostMetadata = HOSTED_APPS_MAP.get(host);
       if (timelineMetricHostMetadata != null && timelineMetricHostMetadata.getUuid() != null) {
-        uuidHostMap.put(new String(timelineMetricHostMetadata.getUuid()), host);
+        uuidHostMap.put(new TimelineMetricUuid(timelineMetricHostMetadata.getUuid()), host);
       }
     }
   }
@@ -353,13 +411,7 @@
    * @return the UUID generator of type org.apache.ambari.metrics.core.timeline.uuid.MetricUuidGenStrategy
    */
   private MetricUuidGenStrategy getUuidStrategy(Configuration configuration) {
-    String strategy = configuration.get(TIMELINE_METRICS_UUID_GEN_STRATEGY, "");
-    if ("hash".equalsIgnoreCase(strategy)) {
-      return new HashBasedUuidGenStrategy();
-    } else {
-      //Default
-      return new MD5UuidGenStrategy();
-    }
+    return new Murmur3HashUuidGenStrategy();
   }
 
   /**
@@ -379,14 +431,13 @@
     }
 
     if (!createIfNotPresent) {
-      LOG.warn("UUID not found for " + hostname + ", createIfNotPresent is false");
+      LOG.debug("UUID not found for " + hostname + ", createIfNotPresent is false");
       return null;
     }
 
     byte[] uuid = uuidGenStrategy.computeUuid(hostname, HOSTNAME_UUID_LENGTH);
-    String uuidStr = new String(uuid);
-    if (uuidHostMap.containsKey(uuidStr)) {
-      LOG.error("Duplicate key computed for " + hostname +", Collides with  " + uuidHostMap.get(uuidStr));
+    if (uuidHostMap.containsKey(new TimelineMetricUuid(uuid))) {
+      LOG.error("Duplicate key computed for " + hostname +", Collides with  " + uuidHostMap.get(uuid));
       return null;
     }
 
@@ -395,7 +446,7 @@
       HOSTED_APPS_MAP.put(hostname, timelineMetricHostMetadata);
     }
     timelineMetricHostMetadata.setUuid(uuid);
-    uuidHostMap.put(uuidStr, hostname);
+    uuidHostMap.put(new TimelineMetricUuid(uuid), hostname);
 
     return uuid;
   }
@@ -420,17 +471,16 @@
     }
 
     if (!createIfNotPresent) {
-      LOG.warn("UUID not found for " + key + ", createIfNotPresent is false");
+      LOG.debug("UUID not found for " + key + ", createIfNotPresent is false");
       return null;
     }
 
-    byte[] uuid = uuidGenStrategy.computeUuid(timelineClusterMetric, TIMELINE_METRIC_UUID_LENGTH);
+    byte[] uuidBytes = uuidGenStrategy.computeUuid(timelineClusterMetric, TIMELINE_METRIC_UUID_LENGTH);
 
-    String uuidStr = new String(uuid);
-    if (uuidKeyMap.containsKey(uuidStr) && !uuidKeyMap.get(uuidStr).equals(key)) {
-      TimelineMetricMetadataKey collidingKey = (TimelineMetricMetadataKey)uuidKeyMap.get(uuidStr);
-      LOG.error("Duplicate key " + Arrays.toString(uuid) + "(" + uuid +  ") computed for " + timelineClusterMetric.toString()
-        + ", Collides with  " + collidingKey.toString());
+    TimelineMetricUuid uuid = new TimelineMetricUuid(uuidBytes);
+    if (uuidKeyMap.containsKey(uuid) && !uuidKeyMap.get(uuid).equals(key)) {
+      TimelineMetricMetadataKey collidingKey = uuidKeyMap.get(uuid);
+      LOG.error("Duplicate key " + uuid + " computed for " + timelineClusterMetric + ", Collides with  " + collidingKey);
       return null;
     }
 
@@ -442,10 +492,10 @@
       METADATA_CACHE.put(key, timelineMetricMetadata);
     }
 
-    timelineMetricMetadata.setUuid(uuid);
+    timelineMetricMetadata.setUuid(uuid.uuid);
     timelineMetricMetadata.setIsPersisted(false);
-    uuidKeyMap.put(uuidStr, key);
-    return uuid;
+    uuidKeyMap.put(uuid, key);
+    return uuid.uuid;
   }
 
   /**
@@ -484,14 +534,14 @@
     return metricUuid;
   }
 
-  public String getMetricNameFromUuid(byte[]  uuid) {
+  public String getMetricNameFromUuid(byte[] uuid) {
 
     byte[] metricUuid = uuid;
     if (uuid.length == TIMELINE_METRIC_UUID_LENGTH + HOSTNAME_UUID_LENGTH) {
       metricUuid = ArrayUtils.subarray(uuid, 0, TIMELINE_METRIC_UUID_LENGTH);
     }
 
-    TimelineMetricMetadataKey key = uuidKeyMap.get(new String(metricUuid));
+    TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(metricUuid));
     return key != null ? key.getMetricName() : null;
   }
 
@@ -506,11 +556,11 @@
     }
 
     if (uuid.length == TIMELINE_METRIC_UUID_LENGTH) {
-      TimelineMetricMetadataKey key = uuidKeyMap.get(new String(uuid));
+      TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(uuid));
       return key != null ? new TimelineMetric(key.metricName, null, key.appId, key.instanceId) : null;
     } else {
       byte[] metricUuid = ArrayUtils.subarray(uuid, 0, TIMELINE_METRIC_UUID_LENGTH);
-      TimelineMetricMetadataKey key = uuidKeyMap.get(new String(metricUuid));
+      TimelineMetricMetadataKey key = uuidKeyMap.get(new TimelineMetricUuid(metricUuid));
       if (key == null) {
         LOG.error("TimelineMetricMetadataKey is null for : " + Arrays.toString(uuid));
         return null;
@@ -521,7 +571,7 @@
       timelineMetric.setInstanceId(key.instanceId);
 
       byte[] hostUuid = ArrayUtils.subarray(uuid, TIMELINE_METRIC_UUID_LENGTH, HOSTNAME_UUID_LENGTH + TIMELINE_METRIC_UUID_LENGTH);
-      timelineMetric.setHostName(uuidHostMap.get(new String(hostUuid)));
+      timelineMetric.setHostName(uuidHostMap.get(new TimelineMetricUuid(hostUuid)));
       return timelineMetric;
     }
   }
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/query/PhoenixTransactSQL.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/query/PhoenixTransactSQL.java
index f76933a..e0cc642 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/query/PhoenixTransactSQL.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/query/PhoenixTransactSQL.java
@@ -43,8 +43,11 @@
   /**
    * Create table to store individual metric records.
    */
+
+  public static final String METRICS_RECORD_TABLE_NAME = "METRIC_RECORD_UUID";
+
   public static final String CREATE_METRICS_TABLE_SQL = "CREATE TABLE IF NOT " +
-    "EXISTS METRIC_RECORD_UUID (UUID BINARY(32) NOT NULL, " +
+    "EXISTS " + METRICS_RECORD_TABLE_NAME + " (UUID BINARY(20) NOT NULL, " +
     "SERVER_TIME BIGINT NOT NULL, " +
     "METRIC_SUM DOUBLE, " +
     "METRIC_COUNT UNSIGNED_INT, " +
@@ -83,7 +86,7 @@
 
   public static final String CREATE_METRICS_AGGREGATE_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS %s " +
-      "(UUID BINARY(32) NOT NULL, " +
+      "(UUID BINARY(20) NOT NULL, " +
       "SERVER_TIME BIGINT NOT NULL, " +
       "METRIC_SUM DOUBLE," +
       "METRIC_COUNT UNSIGNED_INT, " +
@@ -155,7 +158,7 @@
 
   public static final String CREATE_HOSTED_APPS_METADATA_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS HOSTED_APPS_METADATA_UUID " +
-      "(HOSTNAME VARCHAR, UUID BINARY(16), APP_IDS VARCHAR, " +
+      "(HOSTNAME VARCHAR, UUID BINARY(4), APP_IDS VARCHAR, " +
       "CONSTRAINT pk PRIMARY KEY (HOSTNAME))" +
       "DATA_BLOCK_ENCODING='%s', COMPRESSION='%s'";
 
@@ -165,9 +168,6 @@
       "CONSTRAINT pk PRIMARY KEY (INSTANCE_ID, HOSTNAME))" +
       "DATA_BLOCK_ENCODING='%s', COMPRESSION='%s'";
 
-  public static final String ALTER_METRICS_METADATA_TABLE =
-    "ALTER TABLE METRICS_METADATA_UUID ADD IF NOT EXISTS IS_WHITELISTED BOOLEAN";
-
   ////////////////////////////////
 
   /**
@@ -442,8 +442,6 @@
   public static final String METRICS_CLUSTER_AGGREGATE_DAILY_V1_TABLE_NAME =
     "METRIC_AGGREGATE_DAILY";
 
-  public static final String METRICS_RECORD_TABLE_NAME = "METRIC_RECORD_UUID";
-
   public static final String METRICS_AGGREGATE_MINUTE_TABLE_NAME =
     "METRIC_RECORD_MINUTE_UUID";
   public static final String METRICS_AGGREGATE_HOURLY_TABLE_NAME =
@@ -459,7 +457,7 @@
   public static final String METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME =
     "METRIC_AGGREGATE_DAILY_UUID";
 
-  public static final Pattern PHOENIX_TABLES_REGEX_PATTERN = Pattern.compile("METRIC_.*_UUID");
+  public static final Pattern PHOENIX_TABLES_REGEX_PATTERN = Pattern.compile("METRIC_.*");
 
   public static final String[] PHOENIX_TABLES = {
     METRICS_RECORD_TABLE_NAME,
@@ -469,7 +467,9 @@
     METRICS_CLUSTER_AGGREGATE_TABLE_NAME,
     METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME,
     METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME,
-    METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME
+    METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME,
+    METRIC_TRANSIENT_TABLE_NAME,
+    CONTAINER_METRICS_TABLE_NAME
   };
 
   public static final String DEFAULT_TABLE_COMPRESSION = "SNAPPY";
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/Murmur3HashUuidGenStrategy.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/Murmur3HashUuidGenStrategy.java
new file mode 100644
index 0000000..af8cee5
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/Murmur3HashUuidGenStrategy.java
@@ -0,0 +1,54 @@
+/**
+ * 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.ambari.metrics.core.timeline.uuid;
+
+import com.google.common.hash.Hashing;
+import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
+import org.apache.commons.lang.StringUtils;
+
+public class Murmur3HashUuidGenStrategy implements MetricUuidGenStrategy{
+
+  /**
+   * Compute Murmur3Hash 16 byte UUID for a Metric-App-Instance.
+   * @param timelineClusterMetric input metric
+   * @param maxLength Max length of returned UUID. (Will always be 16 for this technique)
+   * @return 16 byte UUID.
+   */  @Override
+  public byte[] computeUuid(TimelineClusterMetric timelineClusterMetric, int maxLength) {
+
+    String metricString = timelineClusterMetric.getMetricName() + timelineClusterMetric.getAppId();
+    if (StringUtils.isNotEmpty(timelineClusterMetric.getInstanceId())) {
+      metricString += timelineClusterMetric.getInstanceId();
+    }
+    byte[] metricBytes = metricString.getBytes();
+    return Hashing.murmur3_128().hashBytes(metricBytes).asBytes();
+  }
+
+  /**
+   * Compute Murmur3Hash 4 byte UUID for a String.
+   * @param value String input
+   * @param maxLength Max length of returned UUID. (Will always be 4 for this technique)
+   * @return 4 byte UUID.
+   */
+  @Override
+  public byte[] computeUuid(String value, int maxLength) {
+    byte[] valueBytes = value.getBytes();
+    return Hashing.murmur3_32().hashBytes(valueBytes).asBytes();
+  }
+}
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuid.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuid.java
new file mode 100644
index 0000000..7907ff6
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuid.java
@@ -0,0 +1,55 @@
+/**
+ * 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.ambari.metrics.core.timeline.uuid;
+
+import java.util.Arrays;
+
+public class TimelineMetricUuid {
+  public byte[] uuid;
+
+  public TimelineMetricUuid(byte[] uuid) {
+    this.uuid = uuid;
+  }
+
+  @Override
+  public int hashCode() {
+    return Arrays.hashCode(uuid);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+
+    if (this == o) {
+      return false;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    TimelineMetricUuid that = (TimelineMetricUuid) o;
+
+    return Arrays.equals(this.uuid, that.uuid);
+  }
+
+  @Override
+  public String toString() {
+    return Arrays.toString(uuid);
+  }
+}
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/webapp/TimelineWebServices.java b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/webapp/TimelineWebServices.java
index 9c88b1a..3bcbaf6 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/webapp/TimelineWebServices.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/webapp/TimelineWebServices.java
@@ -144,8 +144,8 @@
       // TODO: Check ACLs for MetricEntity using the TimelineACLManager.
       // TODO: Save owner of the MetricEntity.
 
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Storing metrics: " +
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Storing metrics: " +
           TimelineUtils.dumpTimelineRecordtoJSON(metrics, true));
       }
 
@@ -175,8 +175,8 @@
     }
 
     try {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Storing aggregated metrics: " +
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Storing aggregated metrics: " +
                 TimelineUtils.dumpTimelineRecordtoJSON(metrics, true));
       }
 
@@ -200,8 +200,8 @@
     }
 
     try {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Storing container metrics: " + TimelineUtils
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Storing container metrics: " + TimelineUtils
             .dumpTimelineRecordtoJSON(metrics, true));
       }
 
@@ -250,8 +250,8 @@
   ) {
     init(res);
     try {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Request for metrics => metricNames: " + metricNames + ", " +
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("Request for metrics => metricNames: " + metricNames + ", " +
           "appId: " + appId + ", instanceId: " + instanceId + ", " +
           "hostname: " + hostname + ", startTime: " + startTime + ", " +
           "endTime: " + endTime + ", " +
diff --git a/ambari-metrics-timelineservice/src/main/resources/metrics_def/AMSSMOKETESTFAKE.dat b/ambari-metrics-timelineservice/src/main/resources/metrics_def/AMSSMOKETESTFAKE.dat
new file mode 100644
index 0000000..f5c181a
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/main/resources/metrics_def/AMSSMOKETESTFAKE.dat
@@ -0,0 +1 @@
+AMBARI_METRICS.SmokeTest.FakeMetric
\ No newline at end of file
diff --git a/ambari-metrics-timelineservice/src/main/resources/metrics_def/MASTER_HBASE.dat b/ambari-metrics-timelineservice/src/main/resources/metrics_def/HBASE_MASTER.dat
similarity index 100%
rename from ambari-metrics-timelineservice/src/main/resources/metrics_def/MASTER_HBASE.dat
rename to ambari-metrics-timelineservice/src/main/resources/metrics_def/HBASE_MASTER.dat
diff --git a/ambari-metrics-timelineservice/src/main/resources/metrics_def/SLAVE_HBASE.dat b/ambari-metrics-timelineservice/src/main/resources/metrics_def/HBASE_REGIONSERVER.dat
similarity index 100%
rename from ambari-metrics-timelineservice/src/main/resources/metrics_def/SLAVE_HBASE.dat
rename to ambari-metrics-timelineservice/src/main/resources/metrics_def/HBASE_REGIONSERVER.dat
diff --git a/ambari-metrics-timelineservice/src/main/resources/metrics_def/TIMELINE_METRIC_STORE_WATCHER.dat b/ambari-metrics-timelineservice/src/main/resources/metrics_def/TIMELINE_METRIC_STORE_WATCHER.dat
new file mode 100644
index 0000000..af73d02
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/main/resources/metrics_def/TIMELINE_METRIC_STORE_WATCHER.dat
@@ -0,0 +1 @@
+TimelineMetricStoreWatcher.FakeMetric
\ No newline at end of file
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/AbstractMiniHBaseClusterTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/AbstractMiniHBaseClusterTest.java
index 258054c..26078cb 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/AbstractMiniHBaseClusterTest.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/AbstractMiniHBaseClusterTest.java
@@ -18,6 +18,7 @@
 package org.apache.ambari.metrics.core.timeline;
 
 import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.OUT_OFF_BAND_DATA_TIME_ALLOWANCE;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.UPSERT_METRICS_SQL;
 import static org.apache.phoenix.end2end.ParallelStatsDisabledIT.tearDownMiniCluster;
@@ -100,8 +101,12 @@
     // inits connection, starts mini cluster
     conn = getConnection(getUrl());
 
+    Configuration metricsConf = new Configuration();
+    metricsConf.set(TimelineMetricConfiguration.HBASE_COMPRESSION_SCHEME, "NONE");
+
+    metadataManager = new TimelineMetricMetadataManager(metricsConf, hdb);
+    metadataManager.initializeMetadata();
     hdb.initMetricSchema();
-    metadataManager = new TimelineMetricMetadataManager(new Configuration(), hdb);
     hdb.setMetadataInstance(metadataManager);
   }
 
@@ -206,6 +211,8 @@
     metricsConf.set("timeline.metrics.transient.metric.patterns", "topology%");
     // Unit tests insert values into the future
     metricsConf.setLong(OUT_OFF_BAND_DATA_TIME_ALLOWANCE, 600000);
+    metricsConf.set("timeline.metrics." + METRICS_RECORD_TABLE_NAME + ".durability", "SKIP_WAL");
+    metricsConf.set("timeline.metrics." + METRICS_CLUSTER_AGGREGATE_TABLE_NAME + ".durability", "ASYNC_WAL");
 
     return
       new PhoenixHBaseAccessor(new TimelineMetricConfiguration(new Configuration(), metricsConf),
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/ITPhoenixHBaseAccessor.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/ITPhoenixHBaseAccessor.java
index 65b5a1b..20fbc58 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/ITPhoenixHBaseAccessor.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/ITPhoenixHBaseAccessor.java
@@ -19,11 +19,13 @@
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertTrue;
-import static org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor.DATE_TIERED_COMPACTION_POLICY;
-import static org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor.FIFO_COMPACTION_POLICY_CLASS;
-import static org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor.HSTORE_COMPACTION_CLASS_KEY;
-import static org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor.HSTORE_ENGINE_CLASS;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.DATE_TIERED_COMPACTION_POLICY;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.FIFO_COMPACTION_POLICY_CLASS;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HSTORE_COMPACTION_CLASS_KEY;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.HSTORE_ENGINE_CLASS;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.CONTAINER_METRICS_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_TABLE_NAME;
+import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.PHOENIX_TABLES;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.PHOENIX_TABLES_REGEX_PATTERN;
@@ -42,6 +44,7 @@
 import java.util.Map;
 import java.util.Optional;
 
+import org.apache.commons.lang.ArrayUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
@@ -367,28 +370,30 @@
     precisionValues.put(METRICS_RECORD_TABLE_NAME, precisionTtl);
     f.set(hdb, precisionValues);
 
-    Field f2 = PhoenixHBaseAccessor.class.getDeclaredField("timelineMetricsTablesDurability");
-    f2.setAccessible(true);
-    f2.set(hdb, "ASYNC_WAL");
-
     hdb.initPoliciesAndTTL();
 
     // Verify expected policies are set
     boolean normalizerEnabled = false;
     String precisionTableCompactionPolicy = null;
     String aggregateTableCompactionPolicy = null;
-    boolean tableDurabilitySet  = false;
-    for (int i = 0; i < 10; i++) {
+    boolean precisionTableDurabilitySet  = false;
+    boolean aggregateTableDurabilitySet  = false;
+
+    boolean isComplete = false;
+
+    for (int i = 0; i < 10 && !isComplete; i++) {
       LOG.warn("Policy check retry : " + i);
       for (String tableName : PHOENIX_TABLES) {
         TableName[] tableNames = hBaseAdmin.listTableNames(PHOENIX_TABLES_REGEX_PATTERN, false);
+        TableName[] containerMetricsTableName = hBaseAdmin.listTableNames(CONTAINER_METRICS_TABLE_NAME, false);
+        tableNames = (TableName[]) ArrayUtils.addAll(tableNames, containerMetricsTableName);
+
         Optional<TableName> tableNameOptional = Arrays.stream(tableNames)
           .filter(t -> tableName.equals(t.getNameAsString())).findFirst();
 
         TableDescriptor tableDescriptor = hBaseAdmin.getTableDescriptor(tableNameOptional.get());
         
         normalizerEnabled = tableDescriptor.isNormalizationEnabled();
-        tableDurabilitySet = (Durability.ASYNC_WAL.equals(tableDescriptor.getDurability()));
         if (tableName.equals(METRICS_RECORD_TABLE_NAME)) {
           precisionTableCompactionPolicy = tableDescriptor.getValue(HSTORE_COMPACTION_CLASS_KEY);
         } else {
@@ -398,17 +403,25 @@
         // Best effort for 20 seconds
         if (normalizerEnabled || (precisionTableCompactionPolicy == null && aggregateTableCompactionPolicy == null)) {
           Thread.sleep(2000l);
+        } else {
+          isComplete = true;
         }
         if (tableName.equals(METRICS_RECORD_TABLE_NAME)) {
+          precisionTableDurabilitySet = (Durability.SKIP_WAL.equals(tableDescriptor.getDurability()));
           for (ColumnFamilyDescriptor family : tableDescriptor.getColumnFamilies()) {
             precisionTtl = family.getTimeToLive();
           }
         }
+
+        if (tableName.equals(METRICS_CLUSTER_AGGREGATE_TABLE_NAME)) {
+          aggregateTableDurabilitySet = (Durability.ASYNC_WAL.equals(tableDescriptor.getDurability()));
+        }
       }
     }
 
     Assert.assertFalse("Normalizer disabled.", normalizerEnabled);
-    Assert.assertTrue("Durability Set.", tableDurabilitySet);
+    Assert.assertTrue("METRIC_RECORD_UUID Durability Set.", precisionTableDurabilitySet);
+    Assert.assertTrue("METRIC_AGGREGATE_UUID Durability Set.", aggregateTableDurabilitySet);
     Assert.assertEquals("FIFO compaction policy is set for METRIC_RECORD_UUID.", FIFO_COMPACTION_POLICY_CLASS, precisionTableCompactionPolicy);
     Assert.assertEquals("FIFO compaction policy is set for aggregate tables", DATE_TIERED_COMPACTION_POLICY, aggregateTableCompactionPolicy);
     Assert.assertEquals("Precision TTL value as expected.", 86400, precisionTtl);
@@ -441,7 +454,7 @@
     metric.setExitCode(0);
     List<ContainerMetric> list = Arrays.asList(metric);
     hdb.insertContainerMetrics(list);
-    PreparedStatement stmt = conn.prepareStatement("SELECT * FROM CONTAINER_METRICS_UUID");
+    PreparedStatement stmt = conn.prepareStatement("SELECT * FROM CONTAINER_METRICS");
     ResultSet set = stmt.executeQuery();
     // check each filed is set properly when read back.
     boolean foundRecord = false;
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessorTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessorTest.java
index 9d1b2a4..63ec59e 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessorTest.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessorTest.java
@@ -109,7 +109,7 @@
 
     mockStatic(PhoenixTransactSQL.class);
     PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class);
-    Condition condition = new DefaultCondition(Collections.singletonList(new byte[32]), metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true);
+    Condition condition = new DefaultCondition(Collections.singletonList(new byte[20]), metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true);
     expect(PhoenixTransactSQL.prepareGetMetricsSqlStmt(null, condition)).andReturn(preparedStatementMock).once();
     ResultSet rsMock = EasyMock.createNiceMock(ResultSet.class);
     expect(preparedStatementMock.executeQuery()).andReturn(rsMock);
@@ -138,7 +138,7 @@
 
     mockStatic(PhoenixTransactSQL.class);
     PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class);
-    Condition condition = new DefaultCondition(Collections.singletonList(new byte[32]), metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true);
+    Condition condition = new DefaultCondition(Collections.singletonList(new byte[20]), metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true);
     expect(PhoenixTransactSQL.prepareGetMetricsSqlStmt(null, condition)).andReturn(preparedStatementMock).once();
     ResultSet rsMock = EasyMock.createNiceMock(ResultSet.class);
     RuntimeException runtimeException = EasyMock.createNiceMock(RuntimeException.class);
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputerTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputerTest.java
new file mode 100644
index 0000000..150dac2
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricSplitPointComputerTest.java
@@ -0,0 +1,141 @@
+/**
+ * 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.ambari.metrics.core.timeline;
+
+import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
+import org.apache.ambari.metrics.core.timeline.discovery.TimelineMetricMetadataManager;
+import org.apache.hadoop.conf.Configuration;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS;
+import static org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration.TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS;
+import static org.easymock.EasyMock.anyBoolean;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+
+
+public class TimelineMetricSplitPointComputerTest {
+
+  @Test
+  public void testSplitPointComputationForBasicCluster() {
+
+    /**
+     *  HBase Total heap = 1G.
+     *  HDFS,HBASE,YARN services deployed.
+     */
+    Configuration metricsConfMock = EasyMock.createMock(Configuration.class);
+
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS, "")).
+      andReturn("METRICS_COLLECTOR,AMBARI_SERVER,NAMENODE,RESOURCEMANAGER").once();
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS, "")).
+      andReturn("METRICS_MONITOR,DATANODE,NODEMANAGER,HBASE_REGIONSERVER").once();
+    expect(metricsConfMock.getDouble("hbase_total_heapsize", 1024*1024*1024)).andReturn(1024 * 1024 * 1024.0).once();
+
+    Configuration hbaseConfMock = EasyMock.createMock(Configuration.class);
+    expect(hbaseConfMock.getDouble("hbase.regionserver.global.memstore.upperLimit", 0.3)).andReturn(0.3).once();
+    expect(hbaseConfMock.getDouble("hbase.hregion.memstore.flush.size", 134217728)).andReturn(134217728.0).once();
+
+    TimelineMetricMetadataManager metricMetadataManagerMock = EasyMock.createNiceMock(TimelineMetricMetadataManager.class);
+    expect(metricMetadataManagerMock.getUuid(anyObject(TimelineClusterMetric.class), anyBoolean())).andReturn(new byte[16]);
+
+    replay(metricsConfMock, hbaseConfMock, metricMetadataManagerMock);
+
+    TimelineMetricSplitPointComputer timelineMetricSplitPointComputer = new TimelineMetricSplitPointComputer(metricsConfMock,
+      hbaseConfMock,
+      metricMetadataManagerMock);
+    timelineMetricSplitPointComputer.computeSplitPoints();
+
+    Assert.assertEquals(timelineMetricSplitPointComputer.getPrecisionSplitPoints().size(), 3);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getClusterAggregateSplitPoints().size(), 1);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getHostAggregateSplitPoints().size(), 1);
+  }
+
+  @Test
+  public void testSplitPointComputationForMediumCluster() {
+
+    /**
+     *  HBase Total heap = 8G.
+     *  All services deployed.
+     */
+    Configuration metricsConfMock = EasyMock.createMock(Configuration.class);
+
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS, "")).
+      andReturn("METRICS_COLLECTOR,AMBARI_SERVER,NAMENODE,RESOURCEMANAGER," +
+        "NIMBUS,HIVESERVER2,HIVEMETASTORE,HBASE_MASTER,KAFKA_BROKER").once();
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS, "")).
+      andReturn("METRICS_MONITOR,DATANODE,NODEMANAGER,HBASE_REGIONSERVER").once();
+    expect(metricsConfMock.getDouble("hbase_total_heapsize", 1024*1024*1024)).andReturn(8589934592.0).once();
+
+    Configuration hbaseConfMock = EasyMock.createMock(Configuration.class);
+    expect(hbaseConfMock.getDouble("hbase.regionserver.global.memstore.upperLimit", 0.3)).andReturn(0.3).once();
+    expect(hbaseConfMock.getDouble("hbase.hregion.memstore.flush.size", 134217728)).andReturn(134217728.0).once();
+
+    TimelineMetricMetadataManager metricMetadataManagerMock = EasyMock.createNiceMock(TimelineMetricMetadataManager.class);
+    expect(metricMetadataManagerMock.getUuid(anyObject(TimelineClusterMetric.class), anyBoolean())).andReturn(new byte[16]);
+
+    replay(metricsConfMock, hbaseConfMock, metricMetadataManagerMock);
+
+    TimelineMetricSplitPointComputer timelineMetricSplitPointComputer = new TimelineMetricSplitPointComputer(metricsConfMock,
+      hbaseConfMock,
+      metricMetadataManagerMock);
+    timelineMetricSplitPointComputer.computeSplitPoints();
+
+    Assert.assertEquals(timelineMetricSplitPointComputer.getPrecisionSplitPoints().size(), 6);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getClusterAggregateSplitPoints().size(), 1);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getHostAggregateSplitPoints().size(), 1);
+  }
+
+  @Test
+  public void testSplitPointComputationForLargeCluster() {
+
+    /**
+     *  HBase Total heap = 24G.
+     *  All services deployed.
+     */
+    Configuration metricsConfMock = EasyMock.createMock(Configuration.class);
+
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_MASTER_COMPONENTS, "")).
+      andReturn("METRICS_COLLECTOR,AMBARI_SERVER,NAMENODE,RESOURCEMANAGER," +
+        "NIMBUS,HIVESERVER2,HIVEMETASTORE,HBASE_MASTER,KAFKA_BROKER").once();
+    expect(metricsConfMock.get(TIMELINE_METRIC_INITIAL_CONFIGURED_SLAVE_COMPONENTS, "")).
+      andReturn("METRICS_MONITOR,DATANODE,NODEMANAGER,HBASE_REGIONSERVER").once();
+    expect(metricsConfMock.getDouble("hbase_total_heapsize", 1024*1024*1024)).andReturn(24 * 1024 * 1024 * 1024.0).once();
+
+    Configuration hbaseConfMock = EasyMock.createMock(Configuration.class);
+    expect(hbaseConfMock.getDouble("hbase.regionserver.global.memstore.upperLimit", 0.3)).andReturn(0.3).once();
+    expect(hbaseConfMock.getDouble("hbase.hregion.memstore.flush.size", 134217728)).andReturn(2 * 134217728.0).once();
+
+    TimelineMetricMetadataManager metricMetadataManagerMock = EasyMock.createNiceMock(TimelineMetricMetadataManager.class);
+    expect(metricMetadataManagerMock.getUuid(anyObject(TimelineClusterMetric.class), anyBoolean())).andReturn(new byte[16]);
+
+    replay(metricsConfMock, hbaseConfMock, metricMetadataManagerMock);
+
+    TimelineMetricSplitPointComputer timelineMetricSplitPointComputer = new TimelineMetricSplitPointComputer(metricsConfMock,
+      hbaseConfMock,
+      metricMetadataManagerMock);
+    timelineMetricSplitPointComputer.computeSplitPoints();
+
+    Assert.assertEquals(timelineMetricSplitPointComputer.getPrecisionSplitPoints().size(), 14);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getClusterAggregateSplitPoints().size(), 3);
+    Assert.assertEquals(timelineMetricSplitPointComputer.getHostAggregateSplitPoints().size(), 3);
+  }
+}
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcherTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcherTest.java
index de0236c..eb64198 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcherTest.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/TimelineMetricStoreWatcherTest.java
@@ -48,10 +48,10 @@
 
   @Test
   public void testRunPositive() throws Exception {
-    TimelineMetricStore metricStore = createNiceMock(TimelineMetricStore.class);
+    HBaseTimelineMetricsService metricStore = createNiceMock(HBaseTimelineMetricsService.class);
 
-    expect(metricStore.putMetrics(anyObject(TimelineMetrics.class)))
-      .andReturn(new TimelinePutResponse());
+    metricStore.putMetricsSkipCache(anyObject(TimelineMetrics.class));
+    expectLastCall().once();
 
     // metric found
     expect(metricStore.getTimelineMetrics(EasyMock.<List<String>>anyObject(),
@@ -75,7 +75,7 @@
 
   @Test
   public void testRunNegative() throws Exception {
-    TimelineMetricStore metricStore = createNiceMock(TimelineMetricStore.class);
+    HBaseTimelineMetricsService metricStore = createNiceMock(HBaseTimelineMetricsService.class);
 
     expect(metricStore.putMetrics(anyObject(TimelineMetrics.class)))
       .andReturn(new TimelinePutResponse());
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/discovery/TestMetadataManager.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/discovery/TestMetadataManager.java
index 2f2b0b5..28bb75e 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/discovery/TestMetadataManager.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/discovery/TestMetadataManager.java
@@ -160,7 +160,7 @@
 
     byte[] uuid = metadataManager.getUuid(timelineMetric, true);
     Assert.assertNotNull(uuid);
-    Assert.assertEquals(uuid.length, 32);
+    Assert.assertEquals(uuid.length, 20);
 
     byte[] uuidWithoutHost = metadataManager.getUuid(new TimelineClusterMetric(timelineMetric.getMetricName(), timelineMetric.getAppId(), timelineMetric.getInstanceId(), -1), true);
     Assert.assertNotNull(uuidWithoutHost);
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/MetricUuidGenStrategyTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/MetricUuidGenStrategyTest.java
new file mode 100644
index 0000000..a25310b
--- /dev/null
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/MetricUuidGenStrategyTest.java
@@ -0,0 +1,216 @@
+/**
+ * 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.ambari.metrics.core.timeline.uuid;
+
+import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.BufferedReader;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class MetricUuidGenStrategyTest {
+
+
+  private static List<String> apps = Arrays.asList("namenode",
+    "datanode", "hbase_master", "hbase_regionserver", "kafka_broker", "nimbus", "ams-hbase",
+    "accumulo", "nodemanager", "resourcemanager", "ambari_server", "HOST", "timeline_metric_store_watcher",
+    "jobhistoryserver", "hiveserver2", "hivemetastore", "applicationhistoryserver", "amssmoketestfake", "llapdaemon");
+
+  private static Map<String, Set<String>> metricSet  = new HashMap<>();
+
+  @BeforeClass
+  public static void init() {
+    metricSet  = new HashMap<>(populateMetricWhitelistFromFile());
+  }
+
+  @Test
+  @Ignore
+  public void testHashBasedUuid() throws SQLException {
+    testMetricCollisionsForUuidGenStrategy(new HashBasedUuidGenStrategy(), 16);
+  }
+
+  @Test
+  @Ignore
+  public void testHashBasedUuidForAppIds() throws SQLException {
+    MetricUuidGenStrategy strategy = new HashBasedUuidGenStrategy();
+    Map<String, TimelineClusterMetric> uuids = new HashMap<>();
+    for (String app : metricSet.keySet()) {
+      TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric("TestMetric", app, null, -1l);
+      byte[] uuid = strategy.computeUuid(timelineClusterMetric, 16);
+      String uuidStr = new String(uuid);
+      Assert.assertFalse(uuids.containsKey(uuidStr) && !uuids.containsValue(timelineClusterMetric));
+      uuids.put(uuidStr, timelineClusterMetric);
+    }
+  }
+
+  @Test
+  @Ignore
+  public void testHashBasedUuidForHostnames() throws SQLException {
+    testHostCollisionsForUuidGenStrategy(new HashBasedUuidGenStrategy(), 16);
+  }
+
+
+  @Test
+  public void testMD5BasedUuid() throws SQLException {
+    testMetricCollisionsForUuidGenStrategy(new MD5UuidGenStrategy(), 16);
+
+  }
+
+  @Test
+  public void testMD5BasedUuidForHostnames() throws SQLException {
+    testHostCollisionsForUuidGenStrategy(new MD5UuidGenStrategy(), 16);
+  }
+
+
+  @Test
+  public void testMD5ConsistentHashing() throws SQLException, InterruptedException {
+    testConsistencyForUuidGenStrategy(new MD5UuidGenStrategy(), 16);
+  }
+
+
+  @Test
+  public void testMurmur3HashUuid() throws SQLException {
+    testMetricCollisionsForUuidGenStrategy(new Murmur3HashUuidGenStrategy(), 16);
+  }
+
+  @Test
+  public void testMurmur3HashingBasedUuidForHostnames() throws SQLException {
+    testHostCollisionsForUuidGenStrategy(new Murmur3HashUuidGenStrategy(), 4);
+  }
+
+  @Test
+  public void testMurmur3ConsistentHashing() throws SQLException, InterruptedException {
+    testConsistencyForUuidGenStrategy(new Murmur3HashUuidGenStrategy(), 4);
+  }
+
+  private void testMetricCollisionsForUuidGenStrategy(MetricUuidGenStrategy strategy, int uuidLength) {
+    Map<TimelineMetricUuid, TimelineClusterMetric> uuids = new HashMap<>();
+    for (String app : metricSet.keySet()) {
+      Set<String> metrics = metricSet.get(app);
+      for (String m : metrics) {
+        TimelineClusterMetric metric = new TimelineClusterMetric(m, app, null, -1l);
+        byte[] uuid = strategy.computeUuid(metric, uuidLength);
+        Assert.assertNotNull(uuid);
+        Assert.assertTrue(uuid.length == uuidLength);
+        TimelineMetricUuid uuidStr = new TimelineMetricUuid(uuid);
+        Assert.assertFalse(uuids.containsKey(uuidStr) && !uuids.containsValue(metric));
+        uuids.put(uuidStr, metric);
+      }
+    }
+  }
+
+
+  private void testHostCollisionsForUuidGenStrategy(MetricUuidGenStrategy strategy, int uuidLength) {
+    Map<TimelineMetricUuid, String> uuids = new HashMap<>();
+
+    List<String> hosts = new ArrayList<>();
+    String hostPrefix = "TestHost.";
+    String hostSuffix = ".ambari.apache.org";
+
+    for (int i=0; i<=2000; i++) {
+      hosts.add(hostPrefix + i + hostSuffix);
+    }
+
+    for (String host : hosts) {
+      byte[] uuid = strategy.computeUuid(host, uuidLength);
+      Assert.assertNotNull(uuid);
+      Assert.assertTrue(uuid.length == uuidLength);
+      TimelineMetricUuid uuidStr = new TimelineMetricUuid(uuid);
+      Assert.assertFalse(uuids.containsKey(uuidStr));
+      uuids.put(uuidStr, host);
+    }
+  }
+
+  private void testConsistencyForUuidGenStrategy(MetricUuidGenStrategy strategy, int length) throws InterruptedException {
+    String key = "TestString";
+
+    byte[] uuid = strategy.computeUuid(key, length);
+    Assert.assertNotNull(uuid);
+    Assert.assertTrue(uuid.length == length);
+
+    for (int i = 0; i<100; i++) {
+      byte[] uuid2 = strategy.computeUuid(key, length);
+      Assert.assertNotNull(uuid2);
+      Assert.assertTrue(uuid2.length == length);
+      Assert.assertArrayEquals(uuid, uuid2);
+      Thread.sleep(10);
+    }
+  }
+
+  private static Map<String, Set<String>> populateMetricWhitelistFromFile() {
+
+    Map<String, Set<String>> metricSet = new HashMap<String, Set<String>>();
+    FileInputStream fstream = null;
+    Set<String> hbaseMetrics = new HashSet<>();
+    BufferedReader br = null;
+    String strLine;
+    for (String appId : apps) {
+      URL fileUrl = ClassLoader.getSystemResource("metrics_def/" + appId.toUpperCase() + ".dat");
+
+      Set<String> metricsForApp = new HashSet<>();
+      try {
+        fstream = new FileInputStream(fileUrl.getPath());
+        br = new BufferedReader(new InputStreamReader(fstream));
+        while ((strLine = br.readLine()) != null)   {
+          strLine = strLine.trim();
+          metricsForApp.add(strLine);
+        }
+      } catch (Exception ioEx) {
+        System.out.println("Metrics for AppId " + appId + " not found.");
+      } finally {
+        if (br != null) {
+          try {
+            br.close();
+          } catch (IOException e) {
+          }
+        }
+
+        if (fstream != null) {
+          try {
+            fstream.close();
+          } catch (IOException e) {
+          }
+        }
+      }
+      metricsForApp.add("live_hosts");
+      if (appId.startsWith("hbase")) {
+        hbaseMetrics.addAll(metricsForApp);
+      } else {
+        metricSet.put(appId, metricsForApp);
+      }
+      System.out.println("Found " + metricsForApp.size() + " metrics for appId = " + appId);
+    }
+    metricSet.put("hbase", hbaseMetrics);
+    return metricSet;
+  }
+}
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuidManagerTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuidManagerTest.java
deleted file mode 100644
index e4018bb..0000000
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/ambari/metrics/core/timeline/uuid/TimelineMetricUuidManagerTest.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/**
- * 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.ambari.metrics.core.timeline.uuid;
-
-import org.apache.ambari.metrics.core.timeline.aggregators.TimelineClusterMetric;
-import org.junit.Assert;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URL;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-public class TimelineMetricUuidManagerTest {
-
-
-  private List<String> apps = Arrays.asList("namenode",
-    "datanode", "master_hbase", "slave_hbase", "kafka_broker", "nimbus", "ams-hbase",
-    "accumulo", "nodemanager", "resourcemanager", "ambari_server", "HOST", "timeline_metric_store_watcher",
-    "jobhistoryserver", "hiveserver2", "hivemetastore", "applicationhistoryserver", "amssmoketestfake");
-
-  private Map<String, Set<String>> metricSet  = new HashMap<>(populateMetricWhitelistFromFile());
-
-  @Test
-  @Ignore("Collisions possible")
-  public void testHashBasedUuidForMetricName() throws SQLException {
-
-    MetricUuidGenStrategy strategy = new HashBasedUuidGenStrategy();
-    Map<String, TimelineClusterMetric> uuids = new HashMap<>();
-    for (String app : metricSet.keySet()) {
-      Set<String> metrics = metricSet.get(app);
-      for (String metric : metrics) {
-        TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(metric, app, null, -1l);
-        byte[] uuid = strategy.computeUuid(timelineClusterMetric, 16);
-        Assert.assertNotNull(uuid);
-        Assert.assertTrue(uuid.length == 16);
-        String uuidStr = new String(uuid);
-        Assert.assertFalse(uuids.containsKey(uuidStr) && !uuids.containsValue(timelineClusterMetric));
-        uuids.put(uuidStr, timelineClusterMetric);
-      }
-    }
-  }
-
-  @Test
-  public void testHaseBasedUuidForAppIds() throws SQLException {
-
-    MetricUuidGenStrategy strategy = new HashBasedUuidGenStrategy();
-    Map<String, TimelineClusterMetric> uuids = new HashMap<>();
-    for (String app : metricSet.keySet()) {
-      TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric("TestMetric", app, null, -1l);
-      byte[] uuid = strategy.computeUuid(timelineClusterMetric, 16);
-      String uuidStr = new String(uuid);
-      Assert.assertFalse(uuids.containsKey(uuidStr) && !uuids.containsValue(timelineClusterMetric));
-      uuids.put(uuidStr, timelineClusterMetric);
-    }
-  }
-
-  @Test
-  public void testHashBasedUuidForHostnames() throws SQLException {
-
-    MetricUuidGenStrategy strategy = new HashBasedUuidGenStrategy();
-    Map<String, String> uuids = new HashMap<>();
-
-    List<String> hosts = new ArrayList<>();
-    String hostPrefix = "TestHost.";
-    String hostSuffix = ".ambari.apache.org";
-
-    for (int i=0; i<=2000; i++) {
-      hosts.add(hostPrefix + i + hostSuffix);
-    }
-
-    for (String host : hosts) {
-      byte[] uuid = strategy.computeUuid(host, 16);
-      Assert.assertNotNull(uuid);
-      Assert.assertTrue(uuid.length == 16);
-      String uuidStr = new String(uuid);
-      Assert.assertFalse(uuids.containsKey(uuidStr));
-      uuids.put(uuidStr, host);
-    }
-  }
-
-
-  @Test
-  public void testRandomUuidForWhitelistedMetrics() throws SQLException {
-
-    MetricUuidGenStrategy strategy = new MD5UuidGenStrategy();
-    Map<String, String> uuids = new HashMap<>();
-    for (String app : metricSet.keySet()) {
-      Set<String> metrics = metricSet.get(app);
-      for (String metric : metrics) {
-        byte[] uuid = strategy.computeUuid(new TimelineClusterMetric(metric, app, null, -1l), 16);
-        Assert.assertNotNull(uuid);
-        Assert.assertTrue(uuid.length == 16);
-        String uuidStr = new String(uuid);
-        Assert.assertFalse(uuids.containsKey(uuidStr) && !uuids.containsValue(metric));
-        uuids.put(uuidStr, metric);
-      }
-    }
-  }
-
-  @Test
-  public void testRandomUuidForHostnames() throws SQLException {
-
-    MetricUuidGenStrategy strategy = new MD5UuidGenStrategy();
-    Map<String, String> uuids = new HashMap<>();
-
-    List<String> hosts = new ArrayList<>();
-    String hostPrefix = "TestHost.";
-    String hostSuffix = ".ambari.apache.org";
-
-    for (int i=0; i<=2000; i++) {
-      hosts.add(hostPrefix + i + hostSuffix);
-    }
-
-    int numC = 0;
-    for (String host : hosts) {
-      byte[] uuid = strategy.computeUuid(host, 16);
-      Assert.assertNotNull(uuid);
-      Assert.assertTrue(uuid.length == 16);
-      String uuidStr = new String(uuid);
-      Assert.assertFalse(uuids.containsKey(uuidStr));
-      uuids.put(uuidStr, host);
-    }
-  }
-
-
-  @Test
-  public void testConsistentHashing() throws SQLException, InterruptedException {
-
-    MetricUuidGenStrategy strategy = new MD5UuidGenStrategy();
-    String key = "TestString";
-
-    byte[] uuid = strategy.computeUuid(key, 16);
-    Assert.assertNotNull(uuid);
-    Assert.assertTrue(uuid.length == 16);
-
-    for (int i = 0; i<100; i++) {
-      byte[] uuid2 = strategy.computeUuid(key, 16);
-      Assert.assertNotNull(uuid2);
-      Assert.assertTrue(uuid2.length == 16);
-      Assert.assertArrayEquals(uuid, uuid2);
-      Thread.sleep(10);
-    }
-  }
-
-
-  public Map<String, Set<String>> populateMetricWhitelistFromFile() {
-
-
-    Map<String, Set<String>> metricSet = new HashMap<String, Set<String>>();
-    FileInputStream fstream = null;
-    Set<String> hbaseMetrics = new HashSet<>();
-    BufferedReader br = null;
-    String strLine;
-    for (String appId : apps) {
-      URL fileUrl = ClassLoader.getSystemResource("metrics_def/" + appId.toUpperCase() + ".dat");
-
-      Set<String> metricsForApp = new HashSet<>();
-      try {
-        fstream = new FileInputStream(fileUrl.getPath());
-        br = new BufferedReader(new InputStreamReader(fstream));
-        while ((strLine = br.readLine()) != null)   {
-          strLine = strLine.trim();
-          metricsForApp.add(strLine);
-        }
-      } catch (Exception ioEx) {
-        System.out.println("Metrics for AppId " + appId + " not found.");
-      } finally {
-        if (br != null) {
-          try {
-            br.close();
-          } catch (IOException e) {
-          }
-        }
-
-        if (fstream != null) {
-          try {
-            fstream.close();
-          } catch (IOException e) {
-          }
-        }
-      }
-      metricsForApp.add("live_hosts");
-      if (appId.equals("master_hbase") || appId.equals("slave_hbase")) {
-        hbaseMetrics.addAll(metricsForApp);
-      } else {
-        metricSet.put(appId, metricsForApp);
-      }
-      System.out.println("Found " + metricsForApp.size() + " metrics for appId = " + appId);
-    }
-    metricSet.put("hbase", hbaseMetrics);
-    return metricSet;
-  }
-}