Check for task cancelation before fetching exception

[seanstory]

I hit an error:

[FMWK][15:19:19][INFO] [Connector id: confluence, index name: search-confluence, Sync job id: LwAtoJIB52ZH8QDRudeD] Task is canceled, stop Extractor...
--
  | 2024-10-18 15:19:19 UTC | Exception in callback SyncOrchestrator.extractor_task_callback()(<Task cancell.../sink.py:486>>)
  | 2024-10-18 15:19:19 UTC | handle: <Handle SyncOrchestrator.extractor_task_callback()(<Task cancell.../sink.py:486>>)>
  | 2024-10-18 15:19:19 UTC | Traceback (most recent call last):
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/es/sink.py", line 587, in get_docs
  | 2024-10-18 15:19:19 UTC | await lazy_downloads.put(
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/utils.py", line 478, in put
  | 2024-10-18 15:19:19 UTC | await self._sem.acquire()
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/.pyenv/versions/3.11.10/lib/python3.11/asyncio/locks.py", line 387, in acquire
  | 2024-10-18 15:19:19 UTC | await fut
  | 2024-10-18 15:19:19 UTC | asyncio.exceptions.CancelledError
  | 2024-10-18 15:19:19 UTC |  
  | 2024-10-18 15:19:19 UTC | During handling of the above exception, another exception occurred:
  | 2024-10-18 15:19:19 UTC |  
  | 2024-10-18 15:19:19 UTC | Traceback (most recent call last):
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/.pyenv/versions/3.11.10/lib/python3.11/asyncio/events.py", line 84, in _run
  | 2024-10-18 15:19:19 UTC | self._context.run(self._callback, *self._args)
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/es/sink.py", line 1063, in extractor_task_callback
  | 2024-10-18 15:19:19 UTC | if task.exception():
  | 2024-10-18 15:19:19 UTC | ^^^^^^^^^^^^^^^^
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/es/sink.py", line 490, in run
  | 2024-10-18 15:19:19 UTC | await self.get_docs(generator)
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/es/sink.py", line 618, in get_docs
  | 2024-10-18 15:19:19 UTC | await lazy_downloads.join()
  | 2024-10-18 15:19:19 UTC | File "/opt/buildkite-agent/builds/bk-agent-prod-gcp-1729263632898151260/elastic/connectors/connectors/utils.py", line 495, in join
  | 2024-10-18 15:19:19 UTC | await asyncio.gather(*self.tasks, return_exceptions=(not raise_on_error))
  | 2024-10-18 15:19:19 UTC | asyncio.exceptions.CancelledError

Which made me realize that we were not checking in the sink/extractor callbacks for if the task was cancelled before trying to fetch any exception. task.exception() raises a cancelledError if the task was canceled, so it's important to check first.

Checklists

Pre-Review Checklist

• this PR does NOT contain credentials of any kind, such as API keys or username/passwords (double check config.yml.example)
• this PR has a meaningful title
• this PR has a thorough description
• Covered the changes with automated tests
• Tested the changes locally
• Added a label for each target release version (example: v7.13.2, v7.14.0, v8.0.0)

[github-actions]

💚 Backport PR(s) successfully created

Status	Branch	Result
✅	8.16	#2906
✅	8.x	#2907

The backport PRs will be merged automatically after passing CI.