AMBARI-16991 : Fix metric sink. (Miklos Gergely via avijayan)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/fd42b6a2 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/fd42b6a2 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/fd42b6a2 Branch: refs/heads/branch-2.5 Commit: fd42b6a2a7b75667dd63fa7963d41eda2ca56393 Parents: bc9e2d5 Author: Aravindan Vijayan <[email protected]> Authored: Mon Nov 14 21:32:15 2016 -0800 Committer: Aravindan Vijayan <[email protected]> Committed: Tue Nov 15 11:06:39 2016 -0800 ---------------------------------------------------------------------- .../timeline/AbstractTimelineMetricsSink.java | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/fd42b6a2/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java ---------------------------------------------------------------------- diff --git a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java index 57e5589..426eb42 100644 --- a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java +++ b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java @@ -51,10 +51,8 @@ import java.net.URL; import java.security.KeyStore; import java.util.ArrayList; import java.util.Collection; -import java.util.HashSet; import java.util.List; import java.util.Random; -import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -93,7 +91,7 @@ public abstract class AbstractTimelineMetricsSink { // Single element cache with fixed expiration - Helps adjacent Sinks as // well as timed refresh - protected Supplier targetCollectorHostSupplier; + protected Supplier<String> targetCollectorHostSupplier; protected final List<String> allKnownLiveCollectors = new ArrayList<>(); @@ -201,7 +199,7 @@ public abstract class AbstractTimelineMetricsSink { String collectorHost; // Get cached target if (targetCollectorHostSupplier != null) { - collectorHost = (String) targetCollectorHostSupplier.get(); + collectorHost = targetCollectorHostSupplier.get(); // Last X attempts have failed - force refresh if (failedCollectorConnectionsCounter.get() > RETRY_COUNT_BEFORE_COLLECTOR_FAILOVER) { targetCollectorHostSupplier = null; @@ -322,9 +320,9 @@ public abstract class AbstractTimelineMetricsSink { // Auto expire and re-calculate after 1 hour if (targetCollectorHostSupplier != null) { - Object targetCollector = targetCollectorHostSupplier.get(); + String targetCollector = targetCollectorHostSupplier.get(); if (targetCollector != null) { - return (String) targetCollector; + return targetCollector; } } @@ -359,15 +357,15 @@ public abstract class AbstractTimelineMetricsSink { } // Lookup Zookeeper for live hosts - max 10 seconds wait time - if (allKnownLiveCollectors.size() == 0) { + if (allKnownLiveCollectors.size() == 0 && getZookeeperQuorum() != null) { allKnownLiveCollectors.addAll(collectorHAHelper.findLiveCollectorHostsFromZNode()); } if (allKnownLiveCollectors.size() != 0) { targetCollectorHostSupplier = Suppliers.memoizeWithExpiration( - new Supplier() { + new Supplier<String>() { @Override - public Object get() { + public String get() { return metricSinkWriteShardStrategy.findCollectorShard(new ArrayList<>(allKnownLiveCollectors)); } }, // random.nextInt(max - min + 1) + min # (60 to 75 minutes) @@ -377,7 +375,7 @@ public abstract class AbstractTimelineMetricsSink { TimeUnit.MINUTES ); - return (String) targetCollectorHostSupplier.get(); + return targetCollectorHostSupplier.get(); } return null; }
