ACP2E-2427: [TMNA] Many Indexer Log Events per Second

Author: bubasuma

app/code/Magento/Cron/Observer/ProcessCronQueueObserver.php

@@ -337,17 +337,11 @@ private function lockGroup(string $groupId, callable $callback): void
      *
      * @return void
      * @throws Exception|Throwable
+     * @SuppressWarnings(PHPMD.UnusedFormalParameter)
      */
     protected function _runJob($scheduledTime, $currentTime, $jobConfig, $schedule, $groupId)
     {
         $jobCode = $schedule->getJobCode();
-        $scheduleLifetime = $this->getCronGroupConfigurationValue($groupId, self::XML_PATH_SCHEDULE_LIFETIME);
-        $scheduleLifetime = $scheduleLifetime * self::SECONDS_IN_MINUTE;
-        if ($scheduledTime < $currentTime - $scheduleLifetime) {
-            $schedule->setStatus(Schedule::STATUS_MISSED);
-            // phpcs:ignore Magento2.Exceptions.DirectThrow
-            throw new Exception(sprintf('Cron Job %s is missed at %s', $jobCode, $schedule->getScheduledAt()));
-        }
 
         if (!isset($jobConfig['instance'], $jobConfig['method'])) {
             $schedule->setStatus(Schedule::STATUS_ERROR);
@@ -863,7 +857,7 @@ private function processPendingJobs(string $groupId, array $jobsRoot, int $curre
             }
 
             $scheduledTime = strtotime($schedule->getScheduledAt());
-            if ($scheduledTime > $currentTime) {
+            if (!$this->shouldRunJob($schedule, $groupId, $currentTime, (int) $scheduledTime)) {
                 continue;
             }
 
@@ -980,4 +974,62 @@ private function setProcessTitle(string $jobCode, string $groupId): void
             cli_set_process_title($this->originalProcessTitle . " # group: $groupId, job: $jobCode");
         }
     }
+
+    /**
+     * Mark job as missed
+     *
+     * @param Schedule $schedule
+     * @return void
+     */
+    private function markJobAsMissed(Schedule $schedule): void
+    {
+        $jobCode = $schedule->getJobCode();
+        $scheduleId = $schedule->getId();
+        $resource = $schedule->getResource();
+        $connection = $resource->getConnection();
+        $message = sprintf('Cron Job %s is missed at %s', $jobCode, $schedule->getScheduledAt());
+        $result = $this->retrier->execute(
+            function () use ($resource, $connection, $scheduleId, $message) {
+                return $connection->update(
+                    $resource->getTable('cron_schedule'),
+                    ['status' => Schedule::STATUS_MISSED, 'messages' => $message],
+                    ['schedule_id = ?' => $scheduleId, 'status = ?' => Schedule::STATUS_PENDING]
+                );
+            },
+            $connection
+        );
+        if ($result == 1) {
+            $schedule->setStatus(Schedule::STATUS_MISSED);
+            $schedule->setMessages($message);
+            if ($this->state->getMode() === State::MODE_DEVELOPER) {
+                $this->logger->info($message);
+            }
+        }
+    }
+
+    /**
+     * Check if job should be run
+     *
+     * @param Schedule $schedule
+     * @param string $groupId
+     * @param int $currentTime
+     * @param int $scheduledTime
+     * @return bool
+     */
+    private function shouldRunJob(Schedule $schedule, string $groupId, int $currentTime, int $scheduledTime): bool
+    {
+        if ($scheduledTime > $currentTime) {
+            return false;
+        }
+
+        $scheduleLifetime = $this->getCronGroupConfigurationValue($groupId, self::XML_PATH_SCHEDULE_LIFETIME);
+        $scheduleLifetime = $scheduleLifetime * self::SECONDS_IN_MINUTE;
+
+        if ($scheduledTime < $currentTime - $scheduleLifetime) {
+            $this->markJobAsMissed($schedule);
+            return false;
+        }
+
+        return true;
+    }
 }

app/code/Magento/Cron/Test/Unit/Observer/ProcessCronQueueObserverTest.php

@@ -27,6 +27,7 @@
 use Magento\Framework\DB\Adapter\AdapterInterface;
 use Magento\Framework\Event\ManagerInterface;
 use Magento\Framework\Event\Observer;
+use Magento\Framework\Exception\LocalizedException;
 use Magento\Framework\Lock\LockManagerInterface;
 use Magento\Framework\Model\AbstractModel;
 use Magento\Framework\Process\PhpExecutableFinderFactory;
@@ -304,7 +305,17 @@ public function testDispatchCanNotLock(): void
         $this->eventManager->expects($this->never())->method('dispatch');
         $this->cacheMock->expects($this->any())->method('load')->willReturn($lastRun);
         $this->scopeConfigMock->expects($this->any())
-            ->method('getValue')->willReturn(0);
+            ->method('getValue')
+            ->willReturnMap(
+                [
+                    [
+                        'system/cron/test_group/schedule_lifetime',
+                        ScopeInterface::SCOPE_STORE,
+                        null,
+                        2 * 24 * 60
+                    ],
+                ]
+            );
         $this->consoleRequestMock->expects($this->any())
             ->method('getParam')->willReturn('test_group');
 
@@ -366,6 +377,7 @@ public function testDispatchExceptionTooLate(): void
     {
         $exceptionMessage = 'Cron Job test_job1 is missed at 2017-07-30 15:00:00';
         $jobCode = 'test_job1';
+        $scheduleId = 2;
 
         $lastRun = $this->time + 10000000;
         $this->eventManager->expects($this->never())->method('dispatch');
@@ -375,7 +387,7 @@ public function testDispatchExceptionTooLate(): void
 
         $dateScheduledAt = date('Y-m-d H:i:s', $this->time - 86400);
         $schedule = $this->getMockBuilder(Schedule::class)
-            ->onlyMethods(['tryLockJob', 'save', '__wakeup', 'getResource'])
+            ->onlyMethods(['tryLockJob', 'save', '__wakeup', 'getResource', 'getId'])
             ->addMethods(
                 [
                     'getJobCode',
@@ -388,9 +400,10 @@ public function testDispatchExceptionTooLate(): void
                 ]
             )->disableOriginalConstructor()
             ->getMock();
+        $schedule->expects($this->atLeastOnce())->method('getId')->willReturn($scheduleId);
         $schedule->expects($this->atLeastOnce())->method('getJobCode')->willReturn($jobCode);
         $schedule->expects($this->atLeastOnce())->method('getScheduledAt')->willReturn($dateScheduledAt);
-        $schedule->expects($this->once())->method('tryLockJob')->willReturn(true);
+        $schedule->expects($this->never())->method('tryLockJob')->willReturn(true);
         $schedule->expects(
             $this->any()
         )->method(
@@ -399,9 +412,7 @@ public function testDispatchExceptionTooLate(): void
             Schedule::STATUS_MISSED
         )->willReturnSelf();
         $schedule->expects($this->once())->method('setMessages')->with($exceptionMessage);
-        $schedule->expects($this->atLeastOnce())->method('getStatus')->willReturn(Schedule::STATUS_MISSED);
-        $schedule->expects($this->atLeastOnce())->method('getMessages')->willReturn($exceptionMessage);
-        $schedule->expects($this->once())->method('save');
+        $schedule->expects($this->never())->method('save');
         $schedule->expects($this->once())->method('getResource')->willReturn($this->scheduleResourceMock);
 
         $connectionMock = $this->getMockForAbstractClass(AdapterInterface::class);
@@ -410,11 +421,24 @@ public function testDispatchExceptionTooLate(): void
             ->method('getConnection')
             ->willReturn($connectionMock);
 
+        $this->scheduleResourceMock->expects($this->once())
+            ->method('getTable')
+            ->willReturnArgument(0);
+
+        $connectionMock->expects($this->once())
+            ->method('update')
+            ->with(
+                'cron_schedule',
+                ['status' => Schedule::STATUS_MISSED, 'messages' => $exceptionMessage],
+                ['schedule_id = ?' => $scheduleId, 'status = ?' => Schedule::STATUS_PENDING]
+            );
+
         $this->retrierMock->expects($this->once())
             ->method('execute')
             ->willReturnCallback(
                 function ($callback) {
-                    return $callback();
+                    $callback();
+                    return '1';
                 }
             );