def handle(self):
    """Handles kick off request."""
    # Get and verify mr state.
    mr_id = self.request.get("mapreduce_id")
    # Log the mr_id since this is started in an unnamed task
    logging.info("Processing kickoff for job %s", mr_id)
    state = model.MapreduceState.get_by_job_id(mr_id)
    if not self._check_mr_state(state, mr_id):
      return

    # Create input readers.
    readers, serialized_readers_entity = self._get_input_readers(state)
    if readers is None:
      # We don't have any data. Finish map.
      logging.warning("Found no mapper input data to process.")
      state.active = False
      state.result_status = model.MapreduceState.RESULT_SUCCESS
      ControllerCallbackHandler._finalize_job(
          state.mapreduce_spec, state)
      return False

    # Create output writers.
    self._setup_output_writer(state)

    # Save states and make sure we use the saved input readers for
    # subsequent operations.
    result = self._save_states(state, serialized_readers_entity)
    if result is None:
      readers, _ = self._get_input_readers(state)
    elif not result:
      return

    queue_name = self.request.headers.get("X-AppEngine-QueueName")
    KickOffJobHandler._schedule_shards(state.mapreduce_spec, readers,
                                       queue_name,
                                       state.mapreduce_spec.params["base_path"],
                                       state)

    ControllerCallbackHandler.reschedule(
        state, state.mapreduce_spec, serial_id=0, queue_name=queue_name)
Handles kick off request.