@Test public void testStatsAggregationWithAllEmptyNodes() throws IOException { int nodeCount = 3; for (StatsReportType type : EnumSet.of(StatsReportType.ACCOUNT_REPORT, StatsReportType.PARTITION_CLASS_REPORT)) { StatsWrapper emptyNodeStats = generateNodeStats(Collections.emptyList(), DEFAULT_TIMESTAMP, type); String emptyStatsJSON = mapper.writeValueAsString(emptyNodeStats); Map<String, String> instanceToStatsMap = new HashMap<>(); for (int i = 0; i < nodeCount; i++) { instanceToStatsMap.put("Instance_" + i, emptyStatsJSON); } Pair<StatsSnapshot, StatsSnapshot> aggregatedRawAndValidStats = clusterAggregator.doWork(instanceToStatsMap, type); StatsSnapshot expectedSnapshot = new StatsSnapshot(0L, null); StatsSnapshot rawSnapshot = mapper.readValue(mapper.writeValueAsString(aggregatedRawAndValidStats.getFirst()), StatsSnapshot.class); StatsSnapshot validSnapshot = mapper.readValue(mapper.writeValueAsString(aggregatedRawAndValidStats.getSecond()), StatsSnapshot.class); assertTrue("Mismatch in raw snapshot", expectedSnapshot.equals(rawSnapshot)); assertTrue("Mismatch in valid snapshot", expectedSnapshot.equals(validSnapshot)); } }
Pair<StatsSnapshot, StatsSnapshot> doWork(Map<String, String> statsWrappersJSON, StatsReportType type) throws IOException { StatsSnapshot partitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); Map<String, Long> partitionTimestampMap = new HashMap<>(); StatsSnapshot rawPartitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); exceptionOccurredInstances.remove(type); for (Map.Entry<String, String> statsWrapperJSON : statsWrappersJSON.entrySet()) { if (statsWrapperJSON != null && statsWrapperJSON.getValue() != null) { try { StatsWrapper snapshotWrapper = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); StatsWrapper snapshotWrapperCopy = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); combineRawStats(rawPartitionSnapshot, snapshotWrapper); switch (type) { case ACCOUNT_REPORT: combineValidStatsByAccount(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; case PARTITION_CLASS_REPORT: combineValidStatsByPartitionClass(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } catch (Exception e) { logger.error("Exception occurred while processing stats from {}", statsWrapperJSON.getKey(), e); exceptionOccurredInstances.computeIfAbsent(type, key -> new ArrayList<>()).add(statsWrapperJSON.getKey()); } } } if (logger.isTraceEnabled()) { logger.trace("Combined raw snapshot {}", mapper.writeValueAsString(rawPartitionSnapshot)); logger.trace("Combined valid snapshot {}", mapper.writeValueAsString(partitionSnapshot)); } StatsSnapshot reducedRawSnapshot; StatsSnapshot reducedSnapshot; switch (type) { case ACCOUNT_REPORT: reducedRawSnapshot = reduceByAccount(rawPartitionSnapshot); reducedSnapshot = reduceByAccount(partitionSnapshot); break; case PARTITION_CLASS_REPORT: reducedRawSnapshot = reduceByPartitionClass(rawPartitionSnapshot); reducedSnapshot = reduceByPartitionClass(partitionSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } if (logger.isTraceEnabled()) { logger.trace("Reduced raw snapshot {}", mapper.writeValueAsString(reducedRawSnapshot)); logger.trace("Reduced valid snapshot {}", mapper.writeValueAsString(reducedSnapshot)); } return new Pair<>(reducedRawSnapshot, reducedSnapshot); }
HelixClusterAggregator { Pair<StatsSnapshot, StatsSnapshot> doWork(Map<String, String> statsWrappersJSON, StatsReportType type) throws IOException { StatsSnapshot partitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); Map<String, Long> partitionTimestampMap = new HashMap<>(); StatsSnapshot rawPartitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); exceptionOccurredInstances.remove(type); for (Map.Entry<String, String> statsWrapperJSON : statsWrappersJSON.entrySet()) { if (statsWrapperJSON != null && statsWrapperJSON.getValue() != null) { try { StatsWrapper snapshotWrapper = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); StatsWrapper snapshotWrapperCopy = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); combineRawStats(rawPartitionSnapshot, snapshotWrapper); switch (type) { case ACCOUNT_REPORT: combineValidStatsByAccount(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; case PARTITION_CLASS_REPORT: combineValidStatsByPartitionClass(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } catch (Exception e) { logger.error("Exception occurred while processing stats from {}", statsWrapperJSON.getKey(), e); exceptionOccurredInstances.computeIfAbsent(type, key -> new ArrayList<>()).add(statsWrapperJSON.getKey()); } } } if (logger.isTraceEnabled()) { logger.trace("Combined raw snapshot {}", mapper.writeValueAsString(rawPartitionSnapshot)); logger.trace("Combined valid snapshot {}", mapper.writeValueAsString(partitionSnapshot)); } StatsSnapshot reducedRawSnapshot; StatsSnapshot reducedSnapshot; switch (type) { case ACCOUNT_REPORT: reducedRawSnapshot = reduceByAccount(rawPartitionSnapshot); reducedSnapshot = reduceByAccount(partitionSnapshot); break; case PARTITION_CLASS_REPORT: reducedRawSnapshot = reduceByPartitionClass(rawPartitionSnapshot); reducedSnapshot = reduceByPartitionClass(partitionSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } if (logger.isTraceEnabled()) { logger.trace("Reduced raw snapshot {}", mapper.writeValueAsString(reducedRawSnapshot)); logger.trace("Reduced valid snapshot {}", mapper.writeValueAsString(reducedSnapshot)); } return new Pair<>(reducedRawSnapshot, reducedSnapshot); } }
HelixClusterAggregator { Pair<StatsSnapshot, StatsSnapshot> doWork(Map<String, String> statsWrappersJSON, StatsReportType type) throws IOException { StatsSnapshot partitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); Map<String, Long> partitionTimestampMap = new HashMap<>(); StatsSnapshot rawPartitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); exceptionOccurredInstances.remove(type); for (Map.Entry<String, String> statsWrapperJSON : statsWrappersJSON.entrySet()) { if (statsWrapperJSON != null && statsWrapperJSON.getValue() != null) { try { StatsWrapper snapshotWrapper = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); StatsWrapper snapshotWrapperCopy = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); combineRawStats(rawPartitionSnapshot, snapshotWrapper); switch (type) { case ACCOUNT_REPORT: combineValidStatsByAccount(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; case PARTITION_CLASS_REPORT: combineValidStatsByPartitionClass(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } catch (Exception e) { logger.error("Exception occurred while processing stats from {}", statsWrapperJSON.getKey(), e); exceptionOccurredInstances.computeIfAbsent(type, key -> new ArrayList<>()).add(statsWrapperJSON.getKey()); } } } if (logger.isTraceEnabled()) { logger.trace("Combined raw snapshot {}", mapper.writeValueAsString(rawPartitionSnapshot)); logger.trace("Combined valid snapshot {}", mapper.writeValueAsString(partitionSnapshot)); } StatsSnapshot reducedRawSnapshot; StatsSnapshot reducedSnapshot; switch (type) { case ACCOUNT_REPORT: reducedRawSnapshot = reduceByAccount(rawPartitionSnapshot); reducedSnapshot = reduceByAccount(partitionSnapshot); break; case PARTITION_CLASS_REPORT: reducedRawSnapshot = reduceByPartitionClass(rawPartitionSnapshot); reducedSnapshot = reduceByPartitionClass(partitionSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } if (logger.isTraceEnabled()) { logger.trace("Reduced raw snapshot {}", mapper.writeValueAsString(reducedRawSnapshot)); logger.trace("Reduced valid snapshot {}", mapper.writeValueAsString(reducedSnapshot)); } return new Pair<>(reducedRawSnapshot, reducedSnapshot); } HelixClusterAggregator(long relevantTimePeriodInMinutes); }
HelixClusterAggregator { Pair<StatsSnapshot, StatsSnapshot> doWork(Map<String, String> statsWrappersJSON, StatsReportType type) throws IOException { StatsSnapshot partitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); Map<String, Long> partitionTimestampMap = new HashMap<>(); StatsSnapshot rawPartitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); exceptionOccurredInstances.remove(type); for (Map.Entry<String, String> statsWrapperJSON : statsWrappersJSON.entrySet()) { if (statsWrapperJSON != null && statsWrapperJSON.getValue() != null) { try { StatsWrapper snapshotWrapper = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); StatsWrapper snapshotWrapperCopy = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); combineRawStats(rawPartitionSnapshot, snapshotWrapper); switch (type) { case ACCOUNT_REPORT: combineValidStatsByAccount(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; case PARTITION_CLASS_REPORT: combineValidStatsByPartitionClass(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } catch (Exception e) { logger.error("Exception occurred while processing stats from {}", statsWrapperJSON.getKey(), e); exceptionOccurredInstances.computeIfAbsent(type, key -> new ArrayList<>()).add(statsWrapperJSON.getKey()); } } } if (logger.isTraceEnabled()) { logger.trace("Combined raw snapshot {}", mapper.writeValueAsString(rawPartitionSnapshot)); logger.trace("Combined valid snapshot {}", mapper.writeValueAsString(partitionSnapshot)); } StatsSnapshot reducedRawSnapshot; StatsSnapshot reducedSnapshot; switch (type) { case ACCOUNT_REPORT: reducedRawSnapshot = reduceByAccount(rawPartitionSnapshot); reducedSnapshot = reduceByAccount(partitionSnapshot); break; case PARTITION_CLASS_REPORT: reducedRawSnapshot = reduceByPartitionClass(rawPartitionSnapshot); reducedSnapshot = reduceByPartitionClass(partitionSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } if (logger.isTraceEnabled()) { logger.trace("Reduced raw snapshot {}", mapper.writeValueAsString(reducedRawSnapshot)); logger.trace("Reduced valid snapshot {}", mapper.writeValueAsString(reducedSnapshot)); } return new Pair<>(reducedRawSnapshot, reducedSnapshot); } HelixClusterAggregator(long relevantTimePeriodInMinutes);  }
HelixClusterAggregator { Pair<StatsSnapshot, StatsSnapshot> doWork(Map<String, String> statsWrappersJSON, StatsReportType type) throws IOException { StatsSnapshot partitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); Map<String, Long> partitionTimestampMap = new HashMap<>(); StatsSnapshot rawPartitionSnapshot = new StatsSnapshot(0L, new HashMap<>()); exceptionOccurredInstances.remove(type); for (Map.Entry<String, String> statsWrapperJSON : statsWrappersJSON.entrySet()) { if (statsWrapperJSON != null && statsWrapperJSON.getValue() != null) { try { StatsWrapper snapshotWrapper = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); StatsWrapper snapshotWrapperCopy = mapper.readValue(statsWrapperJSON.getValue(), StatsWrapper.class); combineRawStats(rawPartitionSnapshot, snapshotWrapper); switch (type) { case ACCOUNT_REPORT: combineValidStatsByAccount(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; case PARTITION_CLASS_REPORT: combineValidStatsByPartitionClass(partitionSnapshot, snapshotWrapperCopy, statsWrapperJSON.getKey(), partitionTimestampMap); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } } catch (Exception e) { logger.error("Exception occurred while processing stats from {}", statsWrapperJSON.getKey(), e); exceptionOccurredInstances.computeIfAbsent(type, key -> new ArrayList<>()).add(statsWrapperJSON.getKey()); } } } if (logger.isTraceEnabled()) { logger.trace("Combined raw snapshot {}", mapper.writeValueAsString(rawPartitionSnapshot)); logger.trace("Combined valid snapshot {}", mapper.writeValueAsString(partitionSnapshot)); } StatsSnapshot reducedRawSnapshot; StatsSnapshot reducedSnapshot; switch (type) { case ACCOUNT_REPORT: reducedRawSnapshot = reduceByAccount(rawPartitionSnapshot); reducedSnapshot = reduceByAccount(partitionSnapshot); break; case PARTITION_CLASS_REPORT: reducedRawSnapshot = reduceByPartitionClass(rawPartitionSnapshot); reducedSnapshot = reduceByPartitionClass(partitionSnapshot); break; default: throw new IllegalArgumentException("Unrecognized stats report type: " + type); } if (logger.isTraceEnabled()) { logger.trace("Reduced raw snapshot {}", mapper.writeValueAsString(reducedRawSnapshot)); logger.trace("Reduced valid snapshot {}", mapper.writeValueAsString(reducedSnapshot)); } return new Pair<>(reducedRawSnapshot, reducedSnapshot); } HelixClusterAggregator(long relevantTimePeriodInMinutes);   }