code: private KeyValueState<String, CheckPointState> loadCheckpointState(Map conf, TopologyContext ctx) {
          String namespace = ctx.getThisComponentId() + "-" + ctx.getThisTaskId();
          KeyValueState<String, CheckPointState> state =
                  (KeyValueState<String, CheckPointState>) StateFactory.getState(namespace, conf, ctx);
          if (state.get(TX_STATE_KEY) == null) {
              CheckPointState txState = new CheckPointState(-1, CheckPointState.State.COMMITTED);
              state.put(TX_STATE_KEY, txState);
              state.commit();
              LOG.debug("Initialized checkpoint spout state with txState {}", txState);
          } else {
              LOG.debug("Got checkpoint spout state {}", state.get(TX_STATE_KEY));
          }
          return state;
      }
docs: Loads the last saved checkpoint state the from persistent storage.