Record failed ES queue jobs when saving the result fails (don't retry) (

#28629)

x-pack/plugins/reporting/server/lib/esqueue/__tests__/worker.js

@@ -867,6 +867,33 @@ describe('Worker class', function () {
       return sinon.stub(workerWithFailure, '_failJob').returns(Promise.resolve());
     }
 
+    describe('saving output failure', () => {
+      it('should mark the job as failed if saving to ES fails', async () => {
+        const job = {
+          _id: 'shouldSucced',
+          _source: {
+            timeout: 1000,
+            payload: 'test'
+          }
+        };
+
+        sinon.stub(mockQueue.client, 'update').returns(Promise.reject({ statusCode: 413 }));
+
+        const workerFn = function (jobPayload) {
+          return new Promise(function (resolve) {
+            setTimeout(() => resolve(jobPayload), 10);
+          });
+        };
+        const worker = new Worker(mockQueue, 'test', workerFn, defaultWorkerOptions);
+        const failStub = getFailStub(worker);
+
+        await worker._performJob(job);
+        worker.destroy();
+
+        sinon.assert.called(failStub);
+      });
+    });
+
     describe('search failure', function () {
       it('causes _processPendingJobs to reject the Promise', function () {
         sinon.stub(mockQueue.client, 'search').returns(Promise.reject(new Error('test error')));

x-pack/plugins/reporting/server/lib/esqueue/worker.js

@@ -259,6 +259,7 @@ export class Worker extends events.EventEmitter {
           if (err.statusCode === 409) return false;
           this.warn(`Failure saving job output ${job._id}`, err);
           this.emit(constants.EVENT_WORKER_JOB_UPDATE_ERROR, this._formatErrorParams(err, job));
+          return this._failJob(job, (err.message) ? err.message : false);
         });
     }, (jobErr) => {
       if (!jobErr) {