docs: public static void main(String[] args) throws Exception {
      Builder processingGraphBuilder = Builder.newBuilder();
  
      // A KVStreamlet is produced. Each element is a KeyValue object where the key
      // is the impression ID and the user ID is the value.
      Streamlet<AdImpression> impressions = processingGraphBuilder
          .newSource(AdImpression::new);
  
      // A KVStreamlet is produced. Each element is a KeyValue object where the key
      // is the ad ID and the user ID is the value.
      Streamlet<AdClick> clicks = processingGraphBuilder
          .newSource(AdClick::new);
  
      /**
       * Here, the impressions KVStreamlet is joined to the clicks KVStreamlet.
       */
      impressions
          // The join function here essentially provides the reduce function with a streamlet
          // of KeyValue objects where the userId matches across an impression and a click
          // (meaning that the user has clicked on the ad).
          .join(
              // The other streamlet that's being joined to
              clicks,
              // Key extractor for the impressions streamlet
              impression -> impression.getUserId(),
              // Key extractor for the clicks streamlet
              click -> click.getUserId(),
              // Window configuration for the join operation
              WindowConfig.TumblingCountWindow(25),
              // Join type (inner join means that all elements from both streams will be included)
              JoinType.INNER,
              // For each element resulting from the join operation, a value of 1 will be provided
              // if the ad IDs match between the elements (or a value of 0 if they don't).
              (user1, user2) -> (user1.getAdId().equals(user2.getAdId())) ? 1 : 0
          )
          // The reduce function counts the number of ad clicks per user.
          .reduceByKeyAndWindow(
              // Key extractor for the reduce operation
              kv -> String.format("user-%s", kv.getKey().getKey()),
              // Value extractor for the reduce operation
              kv -> kv.getValue(),
              // Window configuration for the reduce operation
              WindowConfig.TumblingCountWindow(50),
              // A running cumulative total is calculated for each key
              (cumulative, incoming) -> cumulative + incoming
          )
          // Finally, the consumer operation provides formatted log output
          .consume(kw -> {
            LOG.info(String.format("(user: %s, clicks: %d)",
                kw.getKey().getKey(),
                kw.getValue()));
          });
  
      Config config = Config.defaultConfig();
  
      // Fetches the topology name from the first command-line argument
      String topologyName = StreamletUtils.getTopologyName(args);
  
      // Finally, the processing graph and configuration are passed to the Runner, which converts
      // the graph into a Heron topology that can be run in a Heron cluster.
      new Runner().run(topologyName, config, processingGraphBuilder);
    }