code: def reset_for_retry(self, output_writer):
      
      self.input_reader = self.initial_input_reader
      self.slice_id = 0
      self.retries += 1
      self.output_writer = output_writer
      self.handler = self.mapreduce_spec.mapper.handler
docs: Reset self for shard retry.
  
      Args:
        output_writer: new output writer that contains new output files.