code: void initializeState(final ListState<byte[]> bucketStates, final ListState<Long> partCounterState) throws Exception {
  
  		initializePartCounter(partCounterState);
  
  		LOG.info("Subtask {} initializing its state (max part counter={}).", subtaskIndex, maxPartCounter);
  
  		initializeActiveBuckets(bucketStates);
  	}
docs: Initializes the state after recovery from a failure.
  
  <p>During this process:
  <ol>
  <li>we set the initial value for part counter to the maximum value used before across all tasks and buckets.
  This guarantees that we do not overwrite valid data,</li>
  <li>we commit any pending files for previous checkpoints (previous to the last successful one from which we restore),</li>
  <li>we resume writing to the previous in-progress file of each bucket, and</li>
  <li>if we receive multiple states for the same bucket, we merge them.</li>
  </ol>
  @param bucketStates the state holding recovered state about active buckets.
  @param partCounterState the state holding the max previously used part counters.
  @throws Exception if anything goes wrong during retrieving the state or restoring/committing of any
  in-progress/pending part files