code: private void maybeAddWatermarkInputs(ComponentCommon common, IRichBolt bolt) {
          if (bolt instanceof WindowedBoltExecutor) {
              Set<String> comps = new HashSet<>();
              for (GlobalStreamId globalStreamId : common.get_inputs().keySet()) {
                  comps.add(globalStreamId.get_componentId());
              }
  
              for (String comp : comps) {
                  common.put_to_inputs(
                          new GlobalStreamId(comp, Common.WATERMARK_STREAM_ID),
                          Grouping.all(new NullStruct()));
              }
          }
      }
docs: Add watermark stream to source components of window bolts