[ResponseOps] Errors during marking tasks as running are not shown in…

… metrics (#191300)

Resolves #184171

## Summary

Errors are not shown in metrics when Elasticsearch returns an error
during `markAsRunning` (changes status from claiming to running)
operation in TaskManager. This PR updates the TaskManager to throw an
error instead of just logging it.

### Checklist

- [ ] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios


### To verify

1. Create an Always Firing rule.
2. Put the below code in the [try block of TaskStore.bulkUpdate
method](https://github.com/elastic/kibana/blob/692b2285176afc75ff5019e5ed651024d9374f91/x-pack/plugins/task_manager/server/task_store.ts#L304)
to mimic markAsRunning
```
      const isMarkAsRunning = docs.some(
        (doc) =>
          doc.taskType === 'alerting:example.always-firing' &&
          doc.status === 'running' &&
          doc.retryAt !== null
      );
      if (isMarkAsRunning) {
        throw SavedObjectsErrorHelpers.decorateEsUnavailableError(new Error('test'));
      }
```
3. Verify that when the above error is thrown, it is reflected in
[metrics
endpoint](http://localhost:5601/api/task_manager/metrics?reset=false)
results.

x-pack/plugins/task_manager/server/task_pool/task_pool.test.ts

@@ -167,7 +167,7 @@ describe('TaskPool', () => {
       expect(shouldNotRun).not.toHaveBeenCalled();
     });
 
-    test('should log when marking a Task as running fails', async () => {
+    test('should log and throw an error when marking a Task as running fails', async () => {
       const pool = new TaskPool({
         capacity$: of(3),
         definitions,
@@ -180,15 +180,15 @@ describe('TaskPool', () => {
         throw new Error(`Mark Task as running has failed miserably`);
       });
 
-      const result = await pool.run([mockTask(), taskFailedToMarkAsRunning, mockTask()]);
+      await expect(
+        pool.run([mockTask(), taskFailedToMarkAsRunning, mockTask()])
+      ).rejects.toThrowError('Mark Task as running has failed miserably');
 
       expect((logger as jest.Mocked<Logger>).error.mock.calls[0]).toMatchInlineSnapshot(`
         Array [
           "Failed to mark Task TaskType \\"shooooo\\" as running: Mark Task as running has failed miserably",
         ]
       `);
-
-      expect(result).toEqual(TaskPoolRunResult.RunningAllClaimedTasks);
     });
 
     test('should log when running a Task fails', async () => {
@@ -549,7 +549,7 @@ describe('TaskPool', () => {
       expect(shouldNotRun).not.toHaveBeenCalled();
     });
 
-    test('should log when marking a Task as running fails', async () => {
+    test('should log and throw an error when marking a Task as running fails', async () => {
       const pool = new TaskPool({
         capacity$: of(6),
         definitions,
@@ -562,15 +562,15 @@ describe('TaskPool', () => {
         throw new Error(`Mark Task as running has failed miserably`);
       });
 
-      const result = await pool.run([mockTask(), taskFailedToMarkAsRunning, mockTask()]);
+      await expect(
+        pool.run([mockTask(), taskFailedToMarkAsRunning, mockTask()])
+      ).rejects.toThrowError('Mark Task as running has failed miserably');
 
       expect((logger as jest.Mocked<Logger>).error.mock.calls[0]).toMatchInlineSnapshot(`
-        Array [
-          "Failed to mark Task TaskType \\"shooooo\\" as running: Mark Task as running has failed miserably",
-        ]
-      `);
-
-      expect(result).toEqual(TaskPoolRunResult.RunningAllClaimedTasks);
+      Array [
+        "Failed to mark Task TaskType \\"shooooo\\" as running: Mark Task as running has failed miserably",
+      ]
+    `);
     });
 
     test('should log when running a Task fails', async () => {

x-pack/plugins/task_manager/server/task_pool/task_pool.ts

@@ -165,7 +165,10 @@ export class TaskPool {
                       message: VERSION_CONFLICT_MESSAGE,
                     })
               )
-              .catch((err) => this.handleFailureOfMarkAsRunning(taskRunner, err));
+              .catch((err) => {
+                this.handleFailureOfMarkAsRunning(taskRunner, err);
+                throw err;
+              });
           })
       );
     }