Do not enqueue jobs that could not be indexed

src/Service/ElasticSearch.php

@@ -61,7 +61,7 @@ public function indexJob(JobInterface $job, string $indexName): Amp\Promise
     public function bulkIndexJobs(array $jobs, string $indexName): Amp\Promise
     {
         return Amp\call(function () use ($jobs, $indexName) {
-            yield from $this->doBulkIndexJobs($jobs, $indexName);
+            return yield from $this->doBulkIndexJobs($jobs, $indexName);
         });
     }
 
@@ -136,7 +136,7 @@ private function doBulkIndexJobs(array $jobs, string $indexName): Generator
             $body[] = ['index' => ['_id' => $job->getUuid()]];
             $body[] = (array)$this->normalizer->normalize($job, 'json');
         }
-        yield $this->client->bulk($body, $indexName);
+        return yield $this->client->bulk($body, $indexName);
     }
 
     /**

src/Service/QueueManager.php

@@ -218,9 +218,30 @@ private function jobExists(string $jobUuid): Promise
     private function processBatch(): \Generator
     {
         $this->logger->debug('Processing batch');
-        yield $this->elasticSearch->bulkIndexJobs($this->batch, $this->flowConfig->getTube());
+        $result = yield $this->elasticSearch->bulkIndexJobs($this->batch, $this->flowConfig->getTube());
+        $successfullyIndexedJobs = $this->batch;
+        $notIndexedJobs = [];
+
+        if ($result['errors'] === true) {
+            $successfullyIndexedJobs = array_filter($this->batch, function ($job) use ($result) {
+                return $result['items'][$job->getUuid()]['index']['status'] === 201;
+            });
+
+            $notIndexedJobs = array_filter($this->batch, function ($job) use ($result) {
+                return $result['items'][$job->getUuid()]['index']['status'] !== 201;
+            });
+        }
+
+        foreach ($notIndexedJobs as $singleJob) {
+            $this->logger->error(
+                sprintf(
+                    'Job with UUID "%s" could not be indexed in ElasticSearch',
+                    $singleJob->getUuid()
+                )
+            );
+        }
 
-        foreach ($this->batch as $singleJob) {
+        foreach ($successfullyIndexedJobs as $singleJob) {
             yield $this->beanstalkClient->put(
                 $singleJob->getUuid(),
                 $singleJob->getTimeout(),

tests/Integration/ElasticSearchIndexingTest.php

@@ -14,10 +14,13 @@
 use Webgriffe\Esb\KernelTestCase;
 use Webgriffe\Esb\Model\ErroredJobEvent;
 use Webgriffe\Esb\Model\Job;
+use Webgriffe\Esb\Model\JobEventInterface;
 use Webgriffe\Esb\Model\ProducedJobEvent;
 use Webgriffe\Esb\Model\ReservedJobEvent;
 use Webgriffe\Esb\Model\WorkedJobEvent;
 use Webgriffe\Esb\TestUtils;
+use Webgriffe\Esb\Unit\Model\DummyJobEvent;
+use function Amp\Http\formatDateHeader;
 
 class ElasticSearchIndexingTest extends KernelTestCase
 {
@@ -172,6 +175,59 @@ function (Job $job) {
         );
     }
 
+    /**
+     * @test
+     */
+    public function itLogsAndSkipsJobsThatCouldNotBeIndexedOntoElasticSearchWithAllEvents()
+    {
+        $producerDir = vfsStream::url('root/producer_dir');
+        $workerFile = vfsStream::url('root/worker.data');
+        self::createKernel([
+            'services' => [
+                DummyFilesystemRepeatProducer::class => ['arguments' => [$producerDir]],
+                DummyFilesystemWorker::class => ['arguments' => [$workerFile]],
+            ],
+            'flows' => [
+                self::FLOW_CODE => [
+                    'description' => 'ElasticSearch Indexing Test Repeat Flow',
+                    'producer' => ['service' => DummyFilesystemRepeatProducer::class],
+                    'worker' => ['service' => DummyFilesystemWorker::class],
+                ]
+            ]
+        ]);
+        mkdir($producerDir);
+        Loop::delay(
+            200,
+            function () use ($producerDir) {
+                touch($producerDir . DIRECTORY_SEPARATOR . 'job1');
+                // TODO: It needs to become a document with more than 1000 fields
+                $veryLargeDocument = 'TODO';
+                file_put_contents($producerDir . DIRECTORY_SEPARATOR . 'job1', $veryLargeDocument);
+                Loop::delay(
+                    200,
+                    function () use ($producerDir) {
+                        touch($producerDir . DIRECTORY_SEPARATOR . 'job2');
+                    }
+                );
+            }
+        );
+        $this->stopWhen(function () {
+            $successLog = array_filter(
+                $this->logHandler()->getRecords(),
+                function ($log) {
+                    return strpos($log['message'], 'Successfully worked a Job') !== false;
+                }
+            );
+            return count($successLog) >= 1;
+        });
+        self::$kernel->boot();
+
+        Promise\wait($this->esClient->refresh());
+        $search = Promise\wait($this->esClient->uriSearchOneIndex(self::FLOW_CODE, ''));
+        $this->assertCount(1, $search['hits']['hits']); // TODO: Make it green
+        // TODO: Add assertions on logs
+    }
+
     private function assertForEachJob(callable $callable, array $jobsData)
     {
         /** @var Serializer $serializer */