@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);
      //noinspection ConstantConditions
      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);
    }
  }
}