code: def _attempt_slice_recovery(self, shard_state, tstate):
      
      mapper_spec = tstate.mapreduce_spec.mapper
      if not (tstate.output_writer and
              tstate.output_writer._supports_slice_recovery(mapper_spec)):
        return self._TASK_DIRECTIVE.PROCEED_TASK
  
      tstate.output_writer = tstate.output_writer._recover(
          tstate.mapreduce_spec, shard_state.shard_number,
          shard_state.retries + 1)
      return self._TASK_DIRECTIVE.RECOVER_SLICE
docs: Recover a slice.
  
      This is run when a slice had been previously attempted and output
      may have been written. If an output writer requires slice recovery,
      we run those logic to remove output duplicates. Otherwise we just retry
      the slice.
  
      If recovery is needed, then the entire slice will be dedicated
      to recovery logic. No data processing will take place. Thus we call
      the slice "recovery slice". This is needed for correctness:
      An output writer instance can be out of sync from its physical
      medium only when the slice dies after acquring the shard lock but before
      committing shard state to db. The worst failure case is when
      shard state failed to commit after the NAMED task for the next slice was
      added. Thus, recovery slice has a special logic to increment current
      slice_id n to n+2. If the task for n+1 had been added, it will be dropped
      because it is behind shard state.
  
      Args:
        shard_state: an instance of Model.ShardState.
        tstate: an instance of Model.TransientShardState.
  
      Returns:
        _TASK_DIRECTIVE.PROCEED_TASK to continue with this retry.
        _TASK_DIRECTIVE.RECOVER_SLICE to recover this slice.
        The next slice will start at the same input as
        this slice but output to a new instance of output writer.
        Combining outputs from all writer instances is up to implementation.