def METHOD_NAME(self, func, *args, **kwargs):
    logger.info(f"executing task {func}(args={args}, kwargs={kwargs})")
    future = futures.Future()
    context = self.get_run_context()
    context["activity_id"] = str(self.nb_activities)
    self.nb_activities += 1
    # Ensure signals ordering
    if isinstance(func, SignalTask):
        self.signals_sent.add(func.name)
    elif isinstance(func, WaitForSignal):
        signal_name = func.signal_name
        if signal_name not in self.signals_sent:
            raise NotImplementedError(
                f"wait_signal({signal_name}) before signal was sent: unsupported by the local executor"
            )
    elif isinstance(func, MarkerTask):
        self._markers.setdefault(func.name, []).append(Marker(func.name, func.details))
    if isinstance(func, Submittable):
        task = func  # *args, **kwargs already resolved.
        task.context = context
        func = getattr(task, "activity", None)
    elif isinstance(func, Activity):
        kwargs["simpleflow_middlewares"] = self.middlewares
        task = ActivityTask(func, context=context, *args, **kwargs)
    elif issubclass(func, Workflow):
        task = WorkflowTask(self, func, *args, **kwargs)
    else:
        raise TypeError(f"invalid type {type(func)} for {func}")
    if isinstance(task, WorkflowTask):
        self.on_new_workflow(task)
    try:
        future._result = task.execute()
        if hasattr(task, "post_execute"):
            task.post_execute()
        state = "completed"
    except Exception:
        exc_type, exc_value, exc_traceback = sys.exc_info()
        tb = traceback.format_tb(exc_traceback)
        task_failed = exceptions.TaskFailed(
            name=getattr(task, "name", "unknown"),
            reason=format_exc(exc_value),
            details=json_dumps(
                {
                    "error": exc_type.__name__,
                    "error_type": format_exc_type(exc_type),
                    "message": str(exc_value),
                    "traceback": tb,
                },
                default=repr,
            ),
        )
        future.set_exception(task_failed)
        logger.exception(f"rescuing exception: {exc_value}")
        if (isinstance(func, Activity) or issubclass_(func, Workflow)) and getattr(func, "raises_on_failure", None):
            raise task_failed
        state = "failed"
    finally:
        if isinstance(task, WorkflowTask):
            self.on_completed_workflow()
        future._state = futures.FINISHED
    if func:
        self._history.add_activity_task(
            func,
            decision_id=None,
            last_state=state,
            activity_id=context["activity_id"],
            input={"args": args, "kwargs": kwargs},
            result=future.result,
        )
    return future