public void insertMetricRecordsWithMetadata()

in ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/PhoenixHBaseAccessor.java [894:942]

41 lines of code
13 McCabe index (conditional complexity)


  public void insertMetricRecordsWithMetadata(TimelineMetricMetadataManager metadataManager,
                                              TimelineMetrics metrics, boolean skipCache) throws SQLException, IOException {
    List<TimelineMetric> timelineMetrics = metrics.getMetrics();
    if (timelineMetrics == null || timelineMetrics.isEmpty()) {
      LOG.debug("Empty metrics insert request.");
      return;
    }
    for (Iterator<TimelineMetric> iterator = timelineMetrics.iterator(); iterator.hasNext();) {

      TimelineMetric tm = iterator.next();

      boolean acceptMetric = TimelineMetricsFilter.acceptMetric(tm);

      if (supportMultipleClusterMetrics && StringUtils.isEmpty(tm.getInstanceId())) {
        tm.setInstanceId(defaultInstanceId);
      }

      // Write to metadata cache on successful write to store
      if (metadataManager != null) {
        metadataManager.putIfModifiedTimelineMetricMetadata(
                metadataManager.createTimelineMetricMetadata(tm, acceptMetric));

        metadataManager.putIfModifiedHostedAppsMetadata(
                tm.getHostName(), tm.getAppId());

        if (!tm.getAppId().equals("FLUME_HANDLER")) {
          metadataManager.putIfModifiedHostedInstanceMetadata(tm.getInstanceId(), tm.getHostName());
        }
      }
      if (!acceptMetric) {
        iterator.remove();
      }
    }

    if  (!skipCache && cacheEnabled) {
      LOG.debug("Adding metrics to cache");
      if (insertCache.size() >= cacheSize) {
        commitMetricsFromCache();
      }
      try {
        insertCache.put(metrics); // blocked while the queue is full
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    } else {
      LOG.debug("Skipping metrics cache");
      commitMetrics(metrics);
    }
  }