code: @SuppressWarnings("unchecked")
      @Override
      public Object commit(long batchId, Object state) {
          List<Object> stateList = (List<Object>) state;
          if (stateOperator != null) {
              Object commitState = stateOperator.commit(batchId, stateList);
              stateList.add(commitState);
          }
          return stateList;
      }
docs: to topology master for persistence