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()) { stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition); rs = stmt.executeQuery(); while (rs.next()) { 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(Configuration hbaseConf,
                              Configuration metricsConf);   PhoenixHBaseAccessor(Configuration hbaseConf,
                       Configuration metricsConf,
                       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 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 saveClusterTimeAggregateRecords(Map<TimelineClusterMetric, MetricHostAggregate> records,
                                              String tableName); boolean isSkipBlockCacheForAggregatorsEnabled(); void saveHostAppsMetadata(Map<String, Set<String>> hostedApps); void saveMetricMetadata(Collection<TimelineMetricMetadata> metricMetadata); Map<String, Set<String>> getHostedAppsMetadata(); Map<TimelineMetricMetadataKey, TimelineMetricMetadata> getTimelineMetricMetadata(); static int RESULTSET_LIMIT; }
@Test public void testGetMetricRecordsPhoenixIOExceptionDoNotRetryException() throws SQLException, IOException { Configuration hbaseConf = new Configuration(); hbaseConf.setStrings(ZOOKEEPER_QUORUM, "quorum"); Configuration metricsConf = new Configuration(); PhoenixConnectionProvider connectionProvider = new PhoenixConnectionProvider() { @Override public HBaseAdmin getHBaseAdmin() throws IOException { return null; } @Override public Connection getConnection() throws SQLException { return null; } }; PhoenixHBaseAccessor accessor = new PhoenixHBaseAccessor(hbaseConf, metricsConf, connectionProvider); List<String> metricNames = new LinkedList<>(); List<String> hostnames = new LinkedList<>(); Multimap<String, List<Function>> metricFunctions = ArrayListMultimap.create(); PowerMock.mockStatic(PhoenixTransactSQL.class); PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class); Condition condition = new DefaultCondition(metricNames, hostnames, "appid", "instanceid", null, null, Precision.SECONDS, 10, true); EasyMock.expect(PhoenixTransactSQL.prepareGetLatestMetricSqlStmt(null, condition)).andReturn(preparedStatementMock).once(); PhoenixTransactSQL.setSortMergeJoinEnabled(true); EasyMock.expectLastCall(); ResultSet rsMock = EasyMock.createNiceMock(ResultSet.class); PhoenixIOException pioe1 = EasyMock.createNiceMock(PhoenixIOException.class); PhoenixIOException pioe2 = EasyMock.createNiceMock(PhoenixIOException.class); DoNotRetryIOException dnrioe = EasyMock.createNiceMock(DoNotRetryIOException.class); EasyMock.expect(preparedStatementMock.executeQuery()).andThrow(pioe1); EasyMock.expect(pioe1.getCause()).andReturn(pioe2).atLeastOnce(); EasyMock.expect(pioe2.getCause()).andReturn(dnrioe).atLeastOnce(); StackTraceElement stackTrace[] = new StackTraceElement[]{new StackTraceElement("HashJoinRegionScanner","method","file",1)}; EasyMock.expect(dnrioe.getStackTrace()).andReturn(stackTrace).atLeastOnce(); PowerMock.replayAll(); EasyMock.replay(preparedStatementMock, rsMock, pioe1, pioe2, dnrioe); try { accessor.getMetricRecords(condition, metricFunctions); fail(); } catch (Exception e) { } PowerMock.verifyAll(); }
@Test public void testGetMetricRecords() throws SQLException, IOException { Configuration hbaseConf = new Configuration(); hbaseConf.setStrings(ZOOKEEPER_QUORUM, "quorum"); Configuration metricsConf = new Configuration(); PhoenixConnectionProvider connectionProvider = new PhoenixConnectionProvider() { @Override public HBaseAdmin getHBaseAdmin() throws IOException { return null; } @Override public Connection getConnection() throws SQLException { return null; } }; PhoenixHBaseAccessor accessor = new PhoenixHBaseAccessor(hbaseConf, metricsConf, connectionProvider); List<String> metricNames = new LinkedList<>(); List<String> hostnames = new LinkedList<>(); Multimap<String, List<Function>> metricFunctions = ArrayListMultimap.create(); PowerMock.mockStatic(PhoenixTransactSQL.class); PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class); Condition condition = new DefaultCondition(metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true); EasyMock.expect(PhoenixTransactSQL.prepareGetMetricsSqlStmt(null, condition)).andReturn(preparedStatementMock).once(); ResultSet rsMock = EasyMock.createNiceMock(ResultSet.class); EasyMock.expect(preparedStatementMock.executeQuery()).andReturn(rsMock); PowerMock.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()); PowerMock.verifyAll(); EasyMock.verify(preparedStatementMock, rsMock); }
@Test public void testGetMetricRecordsIOException() throws SQLException, IOException { Configuration hbaseConf = new Configuration(); hbaseConf.setStrings(ZOOKEEPER_QUORUM, "quorum"); Configuration metricsConf = new Configuration(); PhoenixConnectionProvider connectionProvider = new PhoenixConnectionProvider() { @Override public HBaseAdmin getHBaseAdmin() throws IOException { return null; } @Override public Connection getConnection() throws SQLException { return null; } }; PhoenixHBaseAccessor accessor = new PhoenixHBaseAccessor(hbaseConf, metricsConf, connectionProvider); List<String> metricNames = new LinkedList<>(); List<String> hostnames = new LinkedList<>(); Multimap<String, List<Function>> metricFunctions = ArrayListMultimap.create(); PowerMock.mockStatic(PhoenixTransactSQL.class); PreparedStatement preparedStatementMock = EasyMock.createNiceMock(PreparedStatement.class); Condition condition = new DefaultCondition(metricNames, hostnames, "appid", "instanceid", 123L, 234L, Precision.SECONDS, 10, true); EasyMock.expect(PhoenixTransactSQL.prepareGetMetricsSqlStmt(null, condition)).andReturn(preparedStatementMock).once(); ResultSet rsMock = EasyMock.createNiceMock(ResultSet.class); RuntimeException runtimeException = EasyMock.createNiceMock(RuntimeException.class); IOException io = EasyMock.createNiceMock(IOException.class); EasyMock.expect(preparedStatementMock.executeQuery()).andThrow(runtimeException); EasyMock.expect(runtimeException.getCause()).andReturn(io).atLeastOnce(); StackTraceElement stackTrace[] = new StackTraceElement[]{new StackTraceElement("TimeRange","method","file",1)}; EasyMock.expect(io.getStackTrace()).andReturn(stackTrace).atLeastOnce(); PowerMock.replayAll(); EasyMock.replay(preparedStatementMock, rsMock, io, runtimeException); TimelineMetrics tml = accessor.getMetricRecords(condition, metricFunctions); assertEquals(0, tml.getMetrics().size()); PowerMock.verifyAll(); EasyMock.verify(preparedStatementMock, rsMock, io, runtimeException); }