code: private void addCheckPointInputs(ComponentCommon component) {
          Set<GlobalStreamId> checkPointInputs = new HashSet<>();
          for (GlobalStreamId inputStream : component.get_inputs().keySet()) {
              String sourceId = inputStream.get_componentId();
              if (_spouts.containsKey(sourceId)) {
                  checkPointInputs.add(new GlobalStreamId(CheckpointSpout.CHECKPOINT_COMPONENT_ID, CheckpointSpout.CHECKPOINT_STREAM_ID));
              } else {
                  checkPointInputs.add(new GlobalStreamId(sourceId, CheckpointSpout.CHECKPOINT_STREAM_ID));
              }
          }
          for (GlobalStreamId streamId : checkPointInputs) {
              component.put_to_inputs(streamId, Grouping.all(new NullStruct()));
          }
      }
docs: For bolts that has incoming streams from spouts (the root bolts),
  add checkpoint stream from checkpoint spout to its input. For other bolts,
  add checkpoint stream from the previous bolt to its input.