### Question:
TopWordCountGraph implements Supplier<Graph> { @Override public Graph get() { final Pattern pattern = Pattern.compile("\\s"); SimplePipelineBuilder simplePipelineBuilder = new SimplePipelineBuilder(); final SimplePipelineBuilder.Node counter = simplePipelineBuilder.node( new CounterStatefulOp(), new Hashing<WordContainer>() { @Override public int hash(WordContainer wordContainer) { return wordContainer.hashCode(); } @Override public boolean equals(WordContainer left, WordContainer right) { return left.word().equals(right.word()); } } ); final SimplePipelineBuilder.Node splitter = simplePipelineBuilder.node(new MapOp<String, WordEntry>() { @Override public Class<String> inputClass() { return String.class; } @Override public Stream<WordEntry> apply(String s) { return Arrays.stream(pattern.split(s)).map(WordEntry::new); } }); final SimplePipelineBuilder.Node bucketedTop = simplePipelineBuilder.node( new BucketedTopStatefulOp(2), new Hashing<WordCounter>() { @Override public int hash(WordCounter wordCounter) { return wordCounter.hashCode() % 2; } } ); final SimplePipelineBuilder.Node topTop = simplePipelineBuilder.node( new TopTopStatefulOp(2), new Hashing<WordsTop>() {} ); simplePipelineBuilder.connectToSource(splitter); simplePipelineBuilder.connect(splitter, counter); simplePipelineBuilder.connect(counter, bucketedTop); simplePipelineBuilder.connect(bucketedTop, topTop); simplePipelineBuilder.connectToSink(topTop); return simplePipelineBuilder.build(); }  @Override Graph get();  }### Answer:
@Test(dataProvider = "dataProvider", invocationCount = 10) public void topWordCountTest(SystemConfig.Acking acking) throws InterruptedException { try (final LocalRuntime runtime = new LocalRuntime.Builder().maxElementsInGraph(2) .millisBetweenCommits(500) .acking(acking) .build()) { try (final FlameRuntime.Flame flame = runtime.run(new TopWordCountGraph().get())) { final int lineSize = 20; final int streamSize = 10; final Queue<String> input = Stream.generate(() -> { final String[] words = {"repka", "dedka", "babka", "zhuchka", "vnuchka"}; return new Random().ints(lineSize, 0, words.length).mapToObj(i -> words[i]) .collect(joining(" ")); }).limit(streamSize).collect(Collectors.toCollection(ConcurrentLinkedQueue::new)); final Pattern pattern = Pattern.compile("\\s"); final Map<String, Integer> wordCounts = input.stream() .map(pattern::split) .flatMap(Arrays::stream) .collect(toMap(Function.identity(), o -> 1, Integer::sum)); final AwaitResultConsumer<WordsTop> awaitConsumer = new AwaitResultConsumer<>( lineSize * streamSize ); flame.attachRear("wordCountRear", new AkkaRearType<>(runtime.system(), WordsTop.class)) .forEach(r -> r.addListener(awaitConsumer)); final List<AkkaFront.FrontHandle<String>> handles = flame .attachFront("wordCountFront", new AkkaFrontType<String>(runtime.system())) .collect(Collectors.toList()); applyDataToAllHandlesAsync(input, handles); awaitConsumer.await(200, TimeUnit.SECONDS); final WordsTop actualWordsTop = awaitConsumer.result().skip(lineSize * streamSize - 1).findFirst().get(); final Stream<Map.Entry<String, Integer>> entryStream = wordCounts.entrySet() .stream() .filter(entry -> actualWordsTop.wordCounters().containsKey(entry.getKey())); Assert.assertEquals( actualWordsTop.wordCounters(), entryStream .collect(toMap(Map.Entry::getKey, Map.Entry::getValue)) ); Assert.assertEquals( dewordifiedTop(actualWordsTop.wordCounters()), dewordifiedTop(wordCounts).stream().limit(2).collect(Collectors.toList()) ); Assert.assertEquals(actualWordsTop.wordCounters().size(), 2); } } }