无法使用气流插件运行气流任务

如何解决无法使用气流插件运行气流任务

我有 gcloud composer 气流实例和自定义气流 web 插件，它们在特定端点上运行单独的作业。 Airflow 版本为 1.14（1.14+云端composer）

@rest_api_decorator('/run_task',"POST")
def trigger_tasks(self):
        dag_id = Util.get_argument(request,'dag_id')
        task_id = Util.get_argument(request,'task_id')
        subdir = Util.get_argument(request,'subdir')
        execution_date = parsedate(Util.get_argument(request,'execution_date'))

        dag = self.get_dag(subdir,dag_id)

        task = dag.get_task(task_id=task_id)
        ti = TaskInstance(task,execution_date)
        ti.refresh_from_db()
        ti_list.append(ti)

        executor = get_default_executor()
        executor.start()
        executor.queue_task_instance(
                ti,ignore_all_deps=True)
        executor.heartbeat()
        executor.end()

        result = {
            'dag_id': dag_id,'task_id': task_id_list
        }

        return Util.get_response(result)

上面的清单是 Web 插件端点的示例。这段代码在我的本地实例上使用 celery executor 流畅地工作，但由于某种原因它在云中不起作用。 从日志文件中我可以清楚地看到任务正在被添加到队列中：

2021-04-28T16:17:04.962251133Zairflow-webserver [2021-04-28 16:17:01,054] {base_executor.py:58} INFO - Adding to queue: ['airflow','run','tutorial','print_date','2021-04-23T09:18:11+00:00','-A','--local','--pool','default_pool','-sd','DAGS_FOLDER/

但是有多个错误表明上面的操作没有成功完成：

[2021-04-28 16:23:27,934] {redis.py:363} ERROR - Connection to Redis lost: Retry (0/20) now.@-@{"workflow": "tutorial","task-id": "print_date","execution-date": "2021-04-23T09:18:13+00:00"}

我能够使用 gcloud cli 运行单独的任务，但由于某种原因，类似的代码没有在插件中执行。 我错过了什么吗？

更新

这是工作负载的屏幕截图 [1]：https://i.stack.imgur.com/GHUv6.png