code: def _generate_next_task(self):
          
          if self.consumer_status == ConsumerStatus.INITIALIZING:
              self.current_task_exist = True
              self.task_future = self.executor.submit(consumer_initialize_task, self.processor, self.log_client,
                                                      self.shard_id, self.cursor_position, self.cursor_start_time, self.cursor_end_time)
  
          elif self.consumer_status == ConsumerStatus.PROCESSING:
              if self.last_fetch_log_group is not None:
                  self.checkpoint_tracker.set_cursor(self.last_fetch_log_group.end_cursor)
                  self.current_task_exist = True
  
                  # must deep copy cause some revision will happen
                  last_fetch_log_group = copy.deepcopy(self.last_fetch_log_group)
                  self.last_fetch_log_group = None
  
                  if self.last_fetch_count > 0:
                      self.task_future = self.executor.submit(consumer_process_task, self.processor,
                                                              last_fetch_log_group.fetched_log_group_list,
                                                              self.checkpoint_tracker)
  
          elif self.consumer_status == ConsumerStatus.SHUTTING_DOWN:
              self.current_task_exist = True
              self.logger.info("start to cancel fetch job")
              self.cancel_current_fetch()
              self.task_future = self.executor.submit(consumer_shutdown_task, self.processor, self.checkpoint_tracker)
docs: submit consumer framework defined task
          :return: