[8.16] Apply backpressure to the task poller whenever Elasticsearch requests respond with 503 errors (#196900) (#197705)

# Backport

This will backport the following commits from `main` to `8.16`:
- [Apply backpressure to the task poller whenever Elasticsearch requests
respond with 503 errors
(#196900)](https://github.com/elastic/kibana/pull/196900)

<!--- Backport version: 8.9.8 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Ersin
Erdal","email":"92688503+ersin-erdal@users.noreply.github.com"},"sourceCommit":{"committedDate":"2024-10-23T23:16:45Z","message":"Apply
backpressure to the task poller whenever Elasticsearch requests respond
with 503 errors (#196900)\n\nResolves: #195134\r\n\r\nThis PR adds 503
error check to the error filter of\r\n`createManagedConfiguration`
function, besides the 501 error .\r\nSo it applies backpressure to the
task poller for 503 errors as
well.","sha":"292a7d384e51ca9e76d09f817f583bd0b201d9e0","branchLabelMapping":{"^v9.0.0$":"main","^v8.17.0$":"8.x","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["release_note:skip","Team:ResponseOps","v9.0.0","backport:prev-minor","v8.16.0","v8.17.0"],"number":196900,"url":"https://github.com/elastic/kibana/pull/196900","mergeCommit":{"message":"Apply
backpressure to the task poller whenever Elasticsearch requests respond
with 503 errors (#196900)\n\nResolves: #195134\r\n\r\nThis PR adds 503
error check to the error filter of\r\n`createManagedConfiguration`
function, besides the 501 error .\r\nSo it applies backpressure to the
task poller for 503 errors as
well.","sha":"292a7d384e51ca9e76d09f817f583bd0b201d9e0"}},"sourceBranch":"main","suggestedTargetBranches":["8.16"],"targetPullRequestStates":[{"branch":"main","label":"v9.0.0","labelRegex":"^v9.0.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/196900","number":196900,"mergeCommit":{"message":"Apply
backpressure to the task poller whenever Elasticsearch requests respond
with 503 errors (#196900)\n\nResolves: #195134\r\n\r\nThis PR adds 503
error check to the error filter of\r\n`createManagedConfiguration`
function, besides the 501 error .\r\nSo it applies backpressure to the
task poller for 503 errors as
well.","sha":"292a7d384e51ca9e76d09f817f583bd0b201d9e0"}},{"branch":"8.16","label":"v8.16.0","labelRegex":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"},{"branch":"8.x","label":"v8.17.0","labelRegex":"^v8.17.0$","isSourceBranch":false,"url":"https://github.com/elastic/kibana/pull/197544","number":197544,"state":"MERGED","mergeCommit":{"sha":"e2001cadd96d112108ee60780bf3708d5fa48a14","message":"[8.x]
Apply backpressure to the task poller whenever Elasticsearch requests
respond with 503 errors (#196900) (#197544)\n\n# Backport\n\nThis will
backport the following commits from `main` to `8.x`:\n- [Apply
backpressure to the task poller whenever Elasticsearch requests\nrespond
with 503
errors\n(#196900)](https://github.com/elastic/kibana/pull/196900)\n\n<!---
Backport version: 9.4.3 -->\n\n### Questions ?\nPlease refer to the
[Backport
tool\ndocumentation](https://github.com/sqren/backport)\n\n<!--BACKPORT
[{\"author\":{\"name\":\"Ersin\nErdal\",\"email\":\"92688503+ersin-erdal@users.noreply.github.com\"},\"sourceCommit\":{\"committedDate\":\"2024-10-23T23:16:45Z\",\"message\":\"Apply\nbackpressure
to the task poller whenever Elasticsearch requests respond\nwith 503
errors (#196900)\\n\\nResolves: #195134\\r\\n\\r\\nThis PR adds
503\nerror check to the error filter
of\\r\\n`createManagedConfiguration`\nfunction, besides the 501 error
.\\r\\nSo it applies backpressure to the\ntask poller for 503 errors
as\nwell.\",\"sha\":\"292a7d384e51ca9e76d09f817f583bd0b201d9e0\",\"branchLabelMapping\":{\"^v9.0.0$\":\"main\",\"^v8.17.0$\":\"8.x\",\"^v(\\\\d+).(\\\\d+).\\\\d+$\":\"$1.$2\"}},\"sourcePullRequest\":{\"labels\":[\"release_note:skip\",\"Team:ResponseOps\",\"v9.0.0\",\"backport:prev-minor\"],\"title\":\"Apply\nbackpressure
to the task poller whenever Elasticsearch requests respond\nwith
503\nerrors\",\"number\":196900,\"url\":\"https://github.com/elastic/kibana/pull/196900\",\"mergeCommit\":{\"message\":\"Apply\nbackpressure
to the task poller whenever Elasticsearch requests respond\nwith 503
errors (#196900)\\n\\nResolves: #195134\\r\\n\\r\\nThis PR adds
503\nerror check to the error filter
of\\r\\n`createManagedConfiguration`\nfunction, besides the 501 error
.\\r\\nSo it applies backpressure to the\ntask poller for 503 errors
as\nwell.\",\"sha\":\"292a7d384e51ca9e76d09f817f583bd0b201d9e0\"}},\"sourceBranch\":\"main\",\"suggestedTargetBranches\":[],\"targetPullRequestStates\":[{\"branch\":\"main\",\"label\":\"v9.0.0\",\"branchLabelMappingKey\":\"^v9.0.0$\",\"isSourceBranch\":true,\"state\":\"MERGED\",\"url\":\"https://github.com/elastic/kibana/pull/196900\",\"number\":196900,\"mergeCommit\":{\"message\":\"Apply\nbackpressure
to the task poller whenever Elasticsearch requests respond\nwith 503
errors (#196900)\\n\\nResolves: #195134\\r\\n\\r\\nThis PR adds
503\nerror check to the error filter
of\\r\\n`createManagedConfiguration`\nfunction, besides the 501 error
.\\r\\nSo it applies backpressure to the\ntask poller for 503 errors
as\nwell.\",\"sha\":\"292a7d384e51ca9e76d09f817f583bd0b201d9e0\"}}]}]
BACKPORT-->\n\nCo-authored-by: Ersin Erdal
<92688503+ersin-erdal@users.noreply.github.com>"}}]}] BACKPORT-->

x-pack/plugins/task_manager/server/lib/create_managed_configuration.test.ts

@@ -184,6 +184,17 @@ describe('createManagedConfiguration()', () => {
         expect(subscription).toHaveBeenNthCalledWith(2, 8);
       });
 
+      test('should decrease configuration at the next interval when a 503 error is emitted', async () => {
+        const { subscription, errors$ } = setupScenario(10);
+        errors$.next(SavedObjectsErrorHelpers.createGenericNotFoundEsUnavailableError('a', 'b'));
+        clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
+        expect(subscription).toHaveBeenCalledTimes(1);
+        expect(subscription).toHaveBeenNthCalledWith(1, 10);
+        clock.tick(1);
+        expect(subscription).toHaveBeenCalledTimes(2);
+        expect(subscription).toHaveBeenNthCalledWith(2, 8);
+      });
+
       test('should log a warning when the configuration changes from the starting value', async () => {
         const { errors$ } = setupScenario(10);
         errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
@@ -235,6 +246,17 @@ describe('createManagedConfiguration()', () => {
         expect(subscription).toHaveBeenNthCalledWith(2, 8);
       });
 
+      test('should decrease configuration at the next interval when a 503 error is emitted', async () => {
+        const { subscription, errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
+        errors$.next(SavedObjectsErrorHelpers.createGenericNotFoundEsUnavailableError('a', 'b'));
+        clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
+        expect(subscription).toHaveBeenCalledTimes(1);
+        expect(subscription).toHaveBeenNthCalledWith(1, 10);
+        clock.tick(1);
+        expect(subscription).toHaveBeenCalledTimes(2);
+        expect(subscription).toHaveBeenNthCalledWith(2, 8);
+      });
+
       test('should log a warning when the configuration changes from the starting value', async () => {
         const { errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
         errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
@@ -305,6 +327,16 @@ describe('createManagedConfiguration()', () => {
       expect(subscription).toHaveBeenNthCalledWith(2, 120);
     });
 
+    test('should increase configuration at the next interval when a 503 error is emitted', async () => {
+      const { subscription, errors$ } = setupScenario(100);
+      errors$.next(SavedObjectsErrorHelpers.createGenericNotFoundEsUnavailableError('a', 'b'));
+      clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
+      expect(subscription).toHaveBeenCalledTimes(1);
+      clock.tick(1);
+      expect(subscription).toHaveBeenCalledTimes(2);
+      expect(subscription).toHaveBeenNthCalledWith(2, 120);
+    });
+
     test('should log a warning when the configuration changes from the starting value', async () => {
       const { errors$ } = setupScenario(100);
       errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));

x-pack/plugins/task_manager/server/lib/create_managed_configuration.ts

@@ -161,7 +161,10 @@ function countErrors(errors$: Observable<Error>, countInterval: number): Observa
     interval(countInterval).pipe(map(() => FLUSH_MARKER)),
     errors$.pipe(
       filter(
-        (e) => SavedObjectsErrorHelpers.isTooManyRequestsError(e) || isEsCannotExecuteScriptError(e)
+        (e) =>
+          SavedObjectsErrorHelpers.isTooManyRequestsError(e) ||
+          SavedObjectsErrorHelpers.isEsUnavailableError(e) ||
+          isEsCannotExecuteScriptError(e)
       )
     )
   ).pipe(