Fix SyncCallback crashing issue

[wangyimosquito]

Issue #, if available:

• While setting an deadline alert with SyncCallback in DAG, worker crashed with error Unrecoverable error: KeyError('ti') [celery.worker] loc=worker.py:207. This is due to the sqs_broker expects a hardcoded payload like {"ti": {"id": "uuid", ...}, "type": "ExecuteTask", ...}, but with SyncCallback configured, the payload looks like {"callback": {"id": "uuid", ...}, "type": "ExecuteCallback", ...}

Description of changes:

• The fix is to handle both types of payload.
• Added unit tests for changed function.

Testing:

• Deployed an environment with the fix and configure a simple deadline alert with a Python print SyncCallback. The DAG run was successful, and checked out the worker log, no KeyError was raised and see callback function executed successfully.
• Also tested the fix with custom sync and async callback function.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.