@Test public void testGetMetricRecords() throws SQLException, IOException { List<String> metricNames = Collections.singletonList("M1"); List<String> hostnames = Collections.singletonList("H1"); Multimap<String, List<Function>> metricFunctions = ArrayListMultimap.create(); mockStatic(PhoenixTransactSQL.class); PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class); 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); replayAll(); EasyMock.replay(preparedStatementMock, rsMock); TimelineMetrics tml = accessor.getMetricRecords(condition, metricFunctions); Condition condition2 = new DefaultCondition(metricNames, hostnames, "appid", "instanceid", 234L, 123L, Precision.SECONDS, 10, true); TimelineMetrics tml2 = accessor.getMetricRecords(condition2, metricFunctions); assertEquals(0, tml2.getMetrics().size()); verifyAll(); EasyMock.verify(preparedStatementMock, rsMock); }
@SuppressWarnings("unchecked") public TimelineMetrics getMetricRecords( final Condition condition, Multimap<String, List<Function>> metricFunctions) throws SQLException, IOException { validateConditionIsNotEmpty(condition); Connection conn = getConnection(); PreparedStatement stmt = null; ResultSet rs = null; TimelineMetrics metrics = new TimelineMetrics(); try { if (condition.isPointInTime()){ getLatestMetricRecords(condition, conn, metrics); } else { if (condition.getEndTime() >= condition.getStartTime()) { if (CollectionUtils.isNotEmpty(condition.getUuids())) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } if (CollectionUtils.isNotEmpty(condition.getTransientMetricNames())) { stmt = PhoenixTransactSQL.prepareTransientMetricsSqlStmt(conn, condition); if (stmt != null) { rs = stmt.executeQuery(); while (rs.next()) { TransientMetricReadHelper.appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } } } else { LOG.warn("Skipping metrics query because endTime < startTime"); } } } catch (PhoenixIOException pioe) { Throwable pioe2 = pioe.getCause(); if (pioe2 instanceof PhoenixIOException && pioe2.getCause() instanceof DoNotRetryIOException) { String className = null; for (StackTraceElement ste : pioe2.getCause().getStackTrace()) { className = ste.getClassName(); } if (className != null && className.equals("HashJoinRegionScanner")) { LOG.error("The cache might have expired and have been removed. Try to" + " increase the cache size by setting bigger value for " + "phoenix.coprocessor.maxMetaDataCacheSize in ams-hbase-site config." + " Falling back to sort-merge join algorithm."); PhoenixTransactSQL.setSortMergeJoinEnabled(true); } } throw pioe; } catch (RuntimeException ex) { Throwable io = ex.getCause(); String className = null; if (io != null) { for (StackTraceElement ste : io.getStackTrace()) { className = ste.getClassName(); } } if (className != null && className.equals("TimeRange")) { LOG.debug(io); return new TimelineMetrics(); } else { throw ex; } } finally { if (rs != null) { try { rs.close(); } catch (SQLException e) { } } if (stmt != null) { try { stmt.close(); } catch (SQLException e) { } } if (conn != null) { try { conn.close(); } catch (SQLException sql) { } } } LOG.debug("Metrics records size: " + metrics.getMetrics().size()); return metrics; }
PhoenixHBaseAccessor { @SuppressWarnings("unchecked") public TimelineMetrics getMetricRecords( final Condition condition, Multimap<String, List<Function>> metricFunctions) throws SQLException, IOException { validateConditionIsNotEmpty(condition); Connection conn = getConnection(); PreparedStatement stmt = null; ResultSet rs = null; TimelineMetrics metrics = new TimelineMetrics(); try { if (condition.isPointInTime()){ getLatestMetricRecords(condition, conn, metrics); } else { if (condition.getEndTime() >= condition.getStartTime()) { if (CollectionUtils.isNotEmpty(condition.getUuids())) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } if (CollectionUtils.isNotEmpty(condition.getTransientMetricNames())) { stmt = PhoenixTransactSQL.prepareTransientMetricsSqlStmt(conn, condition); if (stmt != null) { rs = stmt.executeQuery(); while (rs.next()) { TransientMetricReadHelper.appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } } } else { LOG.warn("Skipping metrics query because endTime < startTime"); } } } catch (PhoenixIOException pioe) { Throwable pioe2 = pioe.getCause(); if (pioe2 instanceof PhoenixIOException && pioe2.getCause() instanceof DoNotRetryIOException) { String className = null; for (StackTraceElement ste : pioe2.getCause().getStackTrace()) { className = ste.getClassName(); } if (className != null && className.equals("HashJoinRegionScanner")) { LOG.error("The cache might have expired and have been removed. Try to" + " increase the cache size by setting bigger value for " + "phoenix.coprocessor.maxMetaDataCacheSize in ams-hbase-site config." + " Falling back to sort-merge join algorithm."); PhoenixTransactSQL.setSortMergeJoinEnabled(true); } } throw pioe; } catch (RuntimeException ex) { Throwable io = ex.getCause(); String className = null; if (io != null) { for (StackTraceElement ste : io.getStackTrace()) { className = ste.getClassName(); } } if (className != null && className.equals("TimeRange")) { LOG.debug(io); return new TimelineMetrics(); } else { throw ex; } } finally { if (rs != null) { try { rs.close(); } catch (SQLException e) { } } if (stmt != null) { try { stmt.close(); } catch (SQLException e) { } } if (conn != null) { try { conn.close(); } catch (SQLException sql) { } } } LOG.debug("Metrics records size: " + metrics.getMetrics().size()); return metrics; } }
PhoenixHBaseAccessor { @SuppressWarnings("unchecked") public TimelineMetrics getMetricRecords( final Condition condition, Multimap<String, List<Function>> metricFunctions) throws SQLException, IOException { validateConditionIsNotEmpty(condition); Connection conn = getConnection(); PreparedStatement stmt = null; ResultSet rs = null; TimelineMetrics metrics = new TimelineMetrics(); try { if (condition.isPointInTime()){ getLatestMetricRecords(condition, conn, metrics); } else { if (condition.getEndTime() >= condition.getStartTime()) { if (CollectionUtils.isNotEmpty(condition.getUuids())) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } if (CollectionUtils.isNotEmpty(condition.getTransientMetricNames())) { stmt = PhoenixTransactSQL.prepareTransientMetricsSqlStmt(conn, condition); if (stmt != null) { rs = stmt.executeQuery(); while (rs.next()) { TransientMetricReadHelper.appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } } } else { LOG.warn("Skipping metrics query because endTime < startTime"); } } } catch (PhoenixIOException pioe) { Throwable pioe2 = pioe.getCause(); if (pioe2 instanceof PhoenixIOException && pioe2.getCause() instanceof DoNotRetryIOException) { String className = null; for (StackTraceElement ste : pioe2.getCause().getStackTrace()) { className = ste.getClassName(); } if (className != null && className.equals("HashJoinRegionScanner")) { LOG.error("The cache might have expired and have been removed. Try to" + " increase the cache size by setting bigger value for " + "phoenix.coprocessor.maxMetaDataCacheSize in ams-hbase-site config." + " Falling back to sort-merge join algorithm."); PhoenixTransactSQL.setSortMergeJoinEnabled(true); } } throw pioe; } catch (RuntimeException ex) { Throwable io = ex.getCause(); String className = null; if (io != null) { for (StackTraceElement ste : io.getStackTrace()) { className = ste.getClassName(); } } if (className != null && className.equals("TimeRange")) { LOG.debug(io); return new TimelineMetrics(); } else { throw ex; } } finally { if (rs != null) { try { rs.close(); } catch (SQLException e) { } } if (stmt != null) { try { stmt.close(); } catch (SQLException e) { } } if (conn != null) { try { conn.close(); } catch (SQLException sql) { } } } LOG.debug("Metrics records size: " + metrics.getMetrics().size()); return metrics; } PhoenixHBaseAccessor(PhoenixConnectionProvider dataSource);   PhoenixHBaseAccessor(TimelineMetricConfiguration configuration, PhoenixConnectionProvider dataSource); }
PhoenixHBaseAccessor { @SuppressWarnings("unchecked") public TimelineMetrics getMetricRecords( final Condition condition, Multimap<String, List<Function>> metricFunctions) throws SQLException, IOException { validateConditionIsNotEmpty(condition); Connection conn = getConnection(); PreparedStatement stmt = null; ResultSet rs = null; TimelineMetrics metrics = new TimelineMetrics(); try { if (condition.isPointInTime()){ getLatestMetricRecords(condition, conn, metrics); } else { if (condition.getEndTime() >= condition.getStartTime()) { if (CollectionUtils.isNotEmpty(condition.getUuids())) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } if (CollectionUtils.isNotEmpty(condition.getTransientMetricNames())) { stmt = PhoenixTransactSQL.prepareTransientMetricsSqlStmt(conn, condition); if (stmt != null) { rs = stmt.executeQuery(); while (rs.next()) { TransientMetricReadHelper.appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } } } else { LOG.warn("Skipping metrics query because endTime < startTime"); } } } catch (PhoenixIOException pioe) { Throwable pioe2 = pioe.getCause(); if (pioe2 instanceof PhoenixIOException && pioe2.getCause() instanceof DoNotRetryIOException) { String className = null; for (StackTraceElement ste : pioe2.getCause().getStackTrace()) { className = ste.getClassName(); } if (className != null && className.equals("HashJoinRegionScanner")) { LOG.error("The cache might have expired and have been removed. Try to" + " increase the cache size by setting bigger value for " + "phoenix.coprocessor.maxMetaDataCacheSize in ams-hbase-site config." + " Falling back to sort-merge join algorithm."); PhoenixTransactSQL.setSortMergeJoinEnabled(true); } } throw pioe; } catch (RuntimeException ex) { Throwable io = ex.getCause(); String className = null; if (io != null) { for (StackTraceElement ste : io.getStackTrace()) { className = ste.getClassName(); } } if (className != null && className.equals("TimeRange")) { LOG.debug(io); return new TimelineMetrics(); } else { throw ex; } } finally { if (rs != null) { try { rs.close(); } catch (SQLException e) { } } if (stmt != null) { try { stmt.close(); } catch (SQLException e) { } } if (conn != null) { try { conn.close(); } catch (SQLException sql) { } } } LOG.debug("Metrics records size: " + metrics.getMetrics().size()); return metrics; } PhoenixHBaseAccessor(PhoenixConnectionProvider dataSource);   PhoenixHBaseAccessor(TimelineMetricConfiguration configuration, PhoenixConnectionProvider dataSource); boolean isInsertCacheEmpty(); void commitMetricsFromCache(); void commitMetrics(TimelineMetrics timelineMetrics); void commitMetrics(Collection<TimelineMetrics> timelineMetricsCollection); @SuppressWarnings("unchecked") static TreeMap<Long, Double> readMetricFromJSON(String json); Connection getConnectionRetryingOnException(); Connection getConnection(); void insertContainerMetrics(List<ContainerMetric> metrics); void insertMetricRecordsWithMetadata(TimelineMetricMetadataManager metadataManager,
                                              TimelineMetrics metrics, boolean skipCache); void insertMetricRecords(TimelineMetrics metrics, boolean skipCache); void insertMetricRecords(TimelineMetrics metrics); @SuppressWarnings("unchecked") TimelineMetrics getMetricRecords(
    final Condition condition, Multimap<String, List<Function>> metricFunctions); TimelineMetrics getAggregateMetricRecords(final Condition condition,
      Multimap<String, List<Function>> metricFunctions); void saveHostAggregateRecords(Map<TimelineMetric, MetricHostAggregate> hostAggregateMap,
                                       String phoenixTableName); void saveClusterAggregateRecords(Map<TimelineClusterMetric, MetricClusterAggregate> records); void saveClusterAggregateRecordsSecond(Map<TimelineClusterMetric, MetricHostAggregate> records,
                                                String tableName); boolean isSkipBlockCacheForAggregatorsEnabled(); void saveHostAppsMetadata(Map<String, TimelineMetricHostMetadata> hostMetadata); void saveInstanceHostsMetadata(Map<String, Set<String>> instanceHostsMap); void saveMetricMetadata(Collection<TimelineMetricMetadata> metricMetadata); Map<String, TimelineMetricHostMetadata> getHostedAppsMetadata(); Map<String, Set<String>> getInstanceHostsMetdata(); Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getTimelineMetricMetadata(); List<TimelineMetricMetadata> scanMetricMetadataForWildCardRequest(Collection<String> metricNames,
                                                                           String appId,
                                                                           String instanceId); List<byte[]> scanHostMetadataForWildCardRequest(List<String> hostnames); Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getTimelineMetricMetadataV1(); void setMetadataInstance(TimelineMetricMetadataManager metadataManager); }
PhoenixHBaseAccessor { @SuppressWarnings("unchecked") public TimelineMetrics getMetricRecords( final Condition condition, Multimap<String, List<Function>> metricFunctions) throws SQLException, IOException { validateConditionIsNotEmpty(condition); Connection conn = getConnection(); PreparedStatement stmt = null; ResultSet rs = null; TimelineMetrics metrics = new TimelineMetrics(); try { if (condition.isPointInTime()){ getLatestMetricRecords(condition, conn, metrics); } else { if (condition.getEndTime() >= condition.getStartTime()) { if (CollectionUtils.isNotEmpty(condition.getUuids())) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } if (CollectionUtils.isNotEmpty(condition.getTransientMetricNames())) { stmt = PhoenixTransactSQL.prepareTransientMetricsSqlStmt(conn, condition); if (stmt != null) { rs = stmt.executeQuery(); while (rs.next()) { TransientMetricReadHelper.appendMetricFromResultSet(metrics, condition, metricFunctions, rs); } } } } else { LOG.warn("Skipping metrics query because endTime < startTime"); } } } catch (PhoenixIOException pioe) { Throwable pioe2 = pioe.getCause(); if (pioe2 instanceof PhoenixIOException && pioe2.getCause() instanceof DoNotRetryIOException) { String className = null; for (StackTraceElement ste : pioe2.getCause().getStackTrace()) { className = ste.getClassName(); } if (className != null && className.equals("HashJoinRegionScanner")) { LOG.error("The cache might have expired and have been removed. Try to" + " increase the cache size by setting bigger value for " + "phoenix.coprocessor.maxMetaDataCacheSize in ams-hbase-site config." + " Falling back to sort-merge join algorithm."); PhoenixTransactSQL.setSortMergeJoinEnabled(true); } } throw pioe; } catch (RuntimeException ex) { Throwable io = ex.getCause(); String className = null; if (io != null) { for (StackTraceElement ste : io.getStackTrace()) { className = ste.getClassName(); } } if (className != null && className.equals("TimeRange")) { LOG.debug(io); return new TimelineMetrics(); } else { throw ex; } } finally { if (rs != null) { try { rs.close(); } catch (SQLException e) { } } if (stmt != null) { try { stmt.close(); } catch (SQLException e) { } } if (conn != null) { try { conn.close(); } catch (SQLException sql) { } } } LOG.debug("Metrics records size: " + metrics.getMetrics().size()); return metrics; } PhoenixHBaseAccessor(PhoenixConnectionProvider dataSource);   PhoenixHBaseAccessor(TimelineMetricConfiguration configuration, PhoenixConnectionProvider dataSource); boolean isInsertCacheEmpty(); void commitMetricsFromCache(); void commitMetrics(TimelineMetrics timelineMetrics); void commitMetrics(Collection<TimelineMetrics> timelineMetricsCollection); @SuppressWarnings("unchecked") static TreeMap<Long, Double> readMetricFromJSON(String json); Connection getConnectionRetryingOnException(); Connection getConnection(); void insertContainerMetrics(List<ContainerMetric> metrics); void insertMetricRecordsWithMetadata(TimelineMetricMetadataManager metadataManager,
                                              TimelineMetrics metrics, boolean skipCache); void insertMetricRecords(TimelineMetrics metrics, boolean skipCache); void insertMetricRecords(TimelineMetrics metrics); @SuppressWarnings("unchecked") TimelineMetrics getMetricRecords(
    final Condition condition, Multimap<String, List<Function>> metricFunctions); TimelineMetrics getAggregateMetricRecords(final Condition condition,
      Multimap<String, List<Function>> metricFunctions); void saveHostAggregateRecords(Map<TimelineMetric, MetricHostAggregate> hostAggregateMap,
                                       String phoenixTableName); void saveClusterAggregateRecords(Map<TimelineClusterMetric, MetricClusterAggregate> records); void saveClusterAggregateRecordsSecond(Map<TimelineClusterMetric, MetricHostAggregate> records,
                                                String tableName); boolean isSkipBlockCacheForAggregatorsEnabled(); void saveHostAppsMetadata(Map<String, TimelineMetricHostMetadata> hostMetadata); void saveInstanceHostsMetadata(Map<String, Set<String>> instanceHostsMap); void saveMetricMetadata(Collection<TimelineMetricMetadata> metricMetadata); Map<String, TimelineMetricHostMetadata> getHostedAppsMetadata(); Map<String, Set<String>> getInstanceHostsMetdata(); Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getTimelineMetricMetadata(); List<TimelineMetricMetadata> scanMetricMetadataForWildCardRequest(Collection<String> metricNames,
                                                                           String appId,
                                                                           String instanceId); List<byte[]> scanHostMetadataForWildCardRequest(List<String> hostnames); Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getTimelineMetricMetadataV1(); void setMetadataInstance(TimelineMetricMetadataManager metadataManager); static int RESULTSET_LIMIT; static int hostMinuteAggregatorDataInterval; static int clusterMinuteAggregatorDataInterval; static int clusterSecondAggregatorDataInterval; }