@Test public void testMerge() { SummarizerConfiguration sc = SummarizerConfiguration.builder(VisibilitySummarizer.class) .addOption(MAX_COUNTERS_OPT, "5").build(); VisibilitySummarizer countSum = new VisibilitySummarizer(); String p = COUNTER_STAT_PREFIX; HashMap<String,Long> sm1 = new HashMap<>(); sm1.put(p + "f001", 9L); sm1.put(p + "f002", 4L); sm1.put(p + "f003", 2L); sm1.put(p + "f004", 1L); sm1.put(p + "f005", 19L); sm1.put(EMITTED_STAT, 15L); sm1.put(SEEN_STAT, 5L); sm1.put(DELETES_IGNORED_STAT, 1L); HashMap<String,Long> sm2 = new HashMap<>(); sm2.put(p + "f001", 1L); sm2.put(p + "f002", 2L); sm2.put(p + "f00a", 7L); sm2.put(p + "f00b", 1L); sm2.put(p + "f00c", 17L); sm2.put(EMITTED_STAT, 18L); sm2.put(SEEN_STAT, 6L); sm2.put(DELETES_IGNORED_STAT, 2L); countSum.combiner(sc).merge(sm1, sm2); HashMap<String,Long> expected = new HashMap<>(); expected.put(p + "f001", 10L); expected.put(p + "f002", 6L); expected.put(p + "f005", 19L); expected.put(p + "f00a", 7L); expected.put(p + "f00c", 17L); expected.put(TOO_LONG_STAT, 0L); expected.put(TOO_MANY_STAT, 4L); expected.put(EMITTED_STAT, 18L + 15L); expected.put(SEEN_STAT, 6L + 5L); expected.put(DELETES_IGNORED_STAT, 3L); assertEquals(expected, sm1); sm2.clear(); sm2.put(p + "f001", 19L); sm2.put(p + "f002", 2L); sm2.put(p + "f003", 3L); sm2.put(p + "f00b", 13L); sm2.put(p + "f00c", 2L); sm2.put(TOO_LONG_STAT, 1L); sm2.put(TOO_MANY_STAT, 3L); sm2.put(EMITTED_STAT, 21L); sm2.put(SEEN_STAT, 7L); sm2.put(DELETES_IGNORED_STAT, 5L); countSum.combiner(sc).merge(sm1, sm2); expected.clear(); expected.put(p + "f001", 29L); expected.put(p + "f002", 8L); expected.put(p + "f005", 19L); expected.put(p + "f00b", 13L); expected.put(p + "f00c", 19L); expected.put(TOO_LONG_STAT, 1L); expected.put(TOO_MANY_STAT, 17L); expected.put(EMITTED_STAT, 21L + 18 + 15); expected.put(SEEN_STAT, 7L + 6 + 5); expected.put(DELETES_IGNORED_STAT, 8L); }
@Override public Combiner combiner(SummarizerConfiguration sc) { init(sc); return (summary1, summary2) -> { for (String key : ALL_STATS) { summary1.merge(key, summary2.getOrDefault(key, 0L), Long::sum); } for (Entry<String,Long> entry : summary2.entrySet()) { String k2 = entry.getKey(); Long v2 = entry.getValue(); if (k2.startsWith(COUNTER_STAT_PREFIX)) { summary1.merge(k2, v2, Long::sum); } } if (summary1.size() - ALL_STATS.length > maxCounters) { List<String> keysToRemove = summary1.entrySet().stream().filter(e -> e.getKey().startsWith(COUNTER_STAT_PREFIX)) .sorted((e1, e2) -> Long.compare(e2.getValue(), e1.getValue())) .skip(maxCounters) .map(Entry::getKey).collect(Collectors.toList()); long removedCount = 0; for (String key : keysToRemove) { removedCount += summary1.remove(key); } summary1.merge(TOO_MANY_STAT, removedCount, Long::sum); } }; }
CountingSummarizer implements Summarizer { @Override public Combiner combiner(SummarizerConfiguration sc) { init(sc); return (summary1, summary2) -> { for (String key : ALL_STATS) { summary1.merge(key, summary2.getOrDefault(key, 0L), Long::sum); } for (Entry<String,Long> entry : summary2.entrySet()) { String k2 = entry.getKey(); Long v2 = entry.getValue(); if (k2.startsWith(COUNTER_STAT_PREFIX)) { summary1.merge(k2, v2, Long::sum); } } if (summary1.size() - ALL_STATS.length > maxCounters) { List<String> keysToRemove = summary1.entrySet().stream().filter(e -> e.getKey().startsWith(COUNTER_STAT_PREFIX)) .sorted((e1, e2) -> Long.compare(e2.getValue(), e1.getValue())) .skip(maxCounters) .map(Entry::getKey).collect(Collectors.toList()); long removedCount = 0; for (String key : keysToRemove) { removedCount += summary1.remove(key); } summary1.merge(TOO_MANY_STAT, removedCount, Long::sum); } }; } }
CountingSummarizer implements Summarizer { @Override public Combiner combiner(SummarizerConfiguration sc) { init(sc); return (summary1, summary2) -> { for (String key : ALL_STATS) { summary1.merge(key, summary2.getOrDefault(key, 0L), Long::sum); } for (Entry<String,Long> entry : summary2.entrySet()) { String k2 = entry.getKey(); Long v2 = entry.getValue(); if (k2.startsWith(COUNTER_STAT_PREFIX)) { summary1.merge(k2, v2, Long::sum); } } if (summary1.size() - ALL_STATS.length > maxCounters) { List<String> keysToRemove = summary1.entrySet().stream().filter(e -> e.getKey().startsWith(COUNTER_STAT_PREFIX)) .sorted((e1, e2) -> Long.compare(e2.getValue(), e1.getValue())) .skip(maxCounters) .map(Entry::getKey).collect(Collectors.toList()); long removedCount = 0; for (String key : keysToRemove) { removedCount += summary1.remove(key); } summary1.merge(TOO_MANY_STAT, removedCount, Long::sum); } }; }  }
CountingSummarizer implements Summarizer { @Override public Combiner combiner(SummarizerConfiguration sc) { init(sc); return (summary1, summary2) -> { for (String key : ALL_STATS) { summary1.merge(key, summary2.getOrDefault(key, 0L), Long::sum); } for (Entry<String,Long> entry : summary2.entrySet()) { String k2 = entry.getKey(); Long v2 = entry.getValue(); if (k2.startsWith(COUNTER_STAT_PREFIX)) { summary1.merge(k2, v2, Long::sum); } } if (summary1.size() - ALL_STATS.length > maxCounters) { List<String> keysToRemove = summary1.entrySet().stream().filter(e -> e.getKey().startsWith(COUNTER_STAT_PREFIX)) .sorted((e1, e2) -> Long.compare(e2.getValue(), e1.getValue())) .skip(maxCounters) .map(Entry::getKey).collect(Collectors.toList()); long removedCount = 0; for (String key : keysToRemove) { removedCount += summary1.remove(key); } summary1.merge(TOO_MANY_STAT, removedCount, Long::sum); } }; }  @Override Collector collector(SummarizerConfiguration sc); @Override Combiner combiner(SummarizerConfiguration sc); }
CountingSummarizer implements Summarizer { @Override public Combiner combiner(SummarizerConfiguration sc) { init(sc); return (summary1, summary2) -> { for (String key : ALL_STATS) { summary1.merge(key, summary2.getOrDefault(key, 0L), Long::sum); } for (Entry<String,Long> entry : summary2.entrySet()) { String k2 = entry.getKey(); Long v2 = entry.getValue(); if (k2.startsWith(COUNTER_STAT_PREFIX)) { summary1.merge(k2, v2, Long::sum); } } if (summary1.size() - ALL_STATS.length > maxCounters) { List<String> keysToRemove = summary1.entrySet().stream().filter(e -> e.getKey().startsWith(COUNTER_STAT_PREFIX)) .sorted((e1, e2) -> Long.compare(e2.getValue(), e1.getValue())) .skip(maxCounters) .map(Entry::getKey).collect(Collectors.toList()); long removedCount = 0; for (String key : keysToRemove) { removedCount += summary1.remove(key); } summary1.merge(TOO_MANY_STAT, removedCount, Long::sum); } }; }  @Override Collector collector(SummarizerConfiguration sc); @Override Combiner combiner(SummarizerConfiguration sc); static final String MAX_COUNTERS_OPT; static final String MAX_COUNTER_LEN_OPT; static final String INGNORE_DELETES_OPT; static final String COUNTER_STAT_PREFIX; static final String TOO_MANY_STAT; static final String TOO_LONG_STAT; static final String EMITTED_STAT; static final String DELETES_IGNORED_STAT; static final String SEEN_STAT; static final String MAX_COUNTER_DEFAULT; static final String MAX_CKL_DEFAULT; static final String INGNORE_DELETES_DEFAULT; }