code: def handle(self):
      
      # 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)
docs: Handles kick off request.