docs: def handle(self):
      """Handle request.
  
      This method has to be careful to pass the same ShardState instance to
      its subroutines calls if the calls mutate or read from ShardState.
      Note especially that Context instance caches and updates the ShardState
      instance.
  
      Returns:
        Set HTTP status code and always returns None.
      """
      # Reconstruct basic states.
      self._start_time = self._time()
      shard_id = self.request.headers[util._MR_SHARD_ID_TASK_HEADER]
      mr_id = self.request.headers[util._MR_ID_TASK_HEADER]
      spec = model.MapreduceSpec._get_mapreduce_spec(mr_id)
      shard_state, control = db.get([
          model.ShardState.get_key_by_shard_id(shard_id),
          model.MapreduceControl.get_key_by_job_id(mr_id),
      ])
  
      # Set context before any IO code is called.
      ctx = context.Context(spec, shard_state,
                            task_retry_count=self.task_retry_count())
      context.Context._set(ctx)
  
      # Unmarshall input reader, output writer, and other transient states.
      tstate = model.TransientShardState.from_request(self.request)
  
      # Try acquire a lease on the shard.
      if shard_state:
        is_this_a_retry = shard_state.acquired_once
      task_directive = self._try_acquire_lease(shard_state, tstate)
      if task_directive in (self._TASK_DIRECTIVE.RETRY_TASK,
                            self._TASK_DIRECTIVE.DROP_TASK):
        return self.__return(shard_state, tstate, task_directive)
      assert task_directive == self._TASK_DIRECTIVE.PROCEED_TASK
  
      # Abort shard if received signal.
      if control and control.command == model.MapreduceControl.ABORT:
        task_directive = self._TASK_DIRECTIVE.ABORT_SHARD
        return self.__return(shard_state, tstate, task_directive)
  
      # Retry shard if user disabled slice retry.
      if (is_this_a_retry and
          parameters.config.TASK_MAX_DATA_PROCESSING_ATTEMPTS <= 1):
        task_directive = self._TASK_DIRECTIVE.RETRY_SHARD
        return self.__return(shard_state, tstate, task_directive)
  
      # TODO(user): Find a better way to set these per thread configs.
      # E.g. what if user change it?
      util._set_ndb_cache_policy()
  
      job_config = map_job.JobConfig._to_map_job_config(
          spec,
          os.environ.get("HTTP_X_APPENGINE_QUEUENAME"))
      job_context = map_job_context.JobContext(job_config)
      self.shard_context = map_job_context.ShardContext(job_context, shard_state)
      self.slice_context = map_job_context.SliceContext(self.shard_context,
                                                        shard_state,
                                                        tstate)
      try:
        slice_id = tstate.slice_id
        self._lc_start_slice(tstate, slice_id)
  
        if shard_state.is_input_finished():
          self._lc_end_slice(tstate, slice_id, last_slice=True)
          # Finalize the stream and set status if there's no more input.
          if (tstate.output_writer and
              isinstance(tstate.output_writer, output_writers.OutputWriter)):
            # It's possible that finalization is successful but
            # saving state failed. In this case this shard will retry upon
            # finalization error.
            # TODO(user): make finalize method idempotent!
            tstate.output_writer.finalize(ctx, shard_state)
          shard_state.set_for_success()
          return self.__return(shard_state, tstate, task_directive)
  
        if is_this_a_retry:
          task_directive = self._attempt_slice_recovery(shard_state, tstate)
          if task_directive != self._TASK_DIRECTIVE.PROCEED_TASK:
            return self.__return(shard_state, tstate, task_directive)
  
        last_slice = self._process_inputs(
            tstate.input_reader, shard_state, tstate, ctx)
  
        self._lc_end_slice(tstate, slice_id)
  
        ctx.flush()
  
        if last_slice:
          # We're done processing data but we still need to finalize the output
          # stream. We save this condition in datastore and force a new slice.
          # That way if finalize fails no input data will be retried.
          shard_state.set_input_finished()
      # pylint: disable=broad-except
      except Exception, e:
        logging.warning("Shard %s got error.", shard_state.shard_id)
        logging.error(traceback.format_exc())
  
        # Fail fast.
        if type(e) is errors.FailJobError:
          logging.error("Got FailJobError.")
          task_directive = self._TASK_DIRECTIVE.FAIL_TASK
        else:
          task_directive = self._TASK_DIRECTIVE.RETRY_SLICE
  
      self.__return(shard_state, tstate, task_directive)