code: private IRichBolt maybeAddCheckpointTupleForwarder(IRichBolt bolt) {
          if (hasStatefulBolt && !(bolt instanceof StatefulBoltExecutor)) {
              bolt = new CheckpointTupleForwarder(bolt);
          }
          return bolt;
      }
docs: If the topology has at least one stateful bolt all the non-stateful bolts
  are wrapped in {@link CheckpointTupleForwarder} so that the checkpoint
  tuples can flow through the topology.