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;
    }
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 ; }
Loads the last saved checkpoint state the from persistent storage.