docs: protected void doPrepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector
        collector, Collection<Event<Tuple>> queue) {
      Objects.requireNonNull(topoConf);
      Objects.requireNonNull(context);
      Objects.requireNonNull(collector);
      Objects.requireNonNull(queue);
      this.windowedOutputCollector = new WindowedOutputCollector(collector);
      bolt.prepare(topoConf, context, windowedOutputCollector);
      this.listener = newWindowLifecycleListener();
      this.windowManager = initWindowManager(listener, topoConf, context, queue);
      start();
      LOG.info(String.format("Initialized window manager %s", windowManager));
    }