Dataaudit and search not working when more than one consumers is running

Hi, Oro team

Intro

I noticed problem with search and dataaudit when more than one consumers is running. This happens when the value of the field of the same entity has changed several times when processing more priority messages. As a result of this at the same time in the queue appeared a lot of audit and search messages for the same entity. When consumers finish processing higher priority messages, they will start processing messages for the data audit and search, so exists a non-zero probability of errors, because consumers can acquire the same resource.

For search:

[error] Consuming interrupted by exception. "An exception occurred while executing 'INSERT INTO "oro_search_item" (entity, alias, record_id, title, changed, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)' with params ["Oro\\Bundle\\TaskBundle\\Entity\\Task", "oro_task", 21, "vfvr", 0, "2017-07-11 17:22:25", "2017-07-11 17:22:25"]:

SQLSTATE[23505]: Unique violation: 7 ERROR:  duplicate key value violates unique constraint "idx_entity"
DETAIL:  Key (entity, record_id)=(Oro\Bundle\TaskBundle\Entity\Task, 21) already exists."
[debug] [RejectMessageOnExceptionDbalExtension] Execution was interrupted and message was rejected. oro.596509517eaaa0.60957659

  [Doctrine\DBAL\Exception\UniqueConstraintViolationException]                                                                                                                               
  An exception occurred while executing 'INSERT INTO "oro_search_item" (entity, alias, record_id, title, changed, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)' with params ["Oro\\  
  Bundle\\TaskBundle\\Entity\\Task", "oro_task", 21, "vfvr", 0, "2017-07-11 17:22:25", "2017-07-11 17:22:25"]:                                                                               
  SQLSTATE[23505]: Unique violation: 7 ERROR:  duplicate key value violates unique constraint "idx_entity"                                                                                   
  DETAIL:  Key (entity, record_id)=(Oro\Bundle\TaskBundle\Entity\Task, 21) already exists.                                                                                                   

Exception trace:
 () at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Driver/AbstractPostgreSQLDriver.php:63
 Doctrine\DBAL\Driver\AbstractPostgreSQLDriver->convertException() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/DBALException.php:128
 Doctrine\DBAL\DBALException::driverExceptionDuringQuery() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Connection.php:996
 Doctrine\DBAL\Connection->executeUpdate() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Connection.php:695
 Doctrine\DBAL\Connection->insert() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Engine/Orm/DBALPersisterDriverTrait.php:245
 Oro\Bundle\SearchBundle\Engine\Orm\BaseDriver->processItems() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Engine/Orm/DBALPersisterDriverTrait.php:110
 Oro\Bundle\SearchBundle\Engine\Orm\BaseDriver->flushWrites() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Entity/Repository/SearchIndexRepository.php:106
 Oro\Bundle\SearchBundle\Entity\Repository\SearchIndexRepository->flushWrites() at /var/www/laboro/application/crm/app/cache/dev/appDevDebugProjectContainer.php:118030
 OroBundleSearchBundleEntityRepositorySearchIndexRepository_00000000043de98e000000006715a134ed13ad841c4ead8d44536f0444da0d40->flushWrites() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Engine/OrmIndexer.php:53
 Oro\Bundle\SearchBundle\Engine\OrmIndexer->Oro\Bundle\SearchBundle\Engine\{closure}() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Connection.php:1108
 Doctrine\DBAL\Connection->transactional() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Engine/OrmIndexer.php:55
 Oro\Bundle\SearchBundle\Engine\OrmIndexer->save() at /var/www/laboro/package/platform/src/Oro/Bundle/SearchBundle/Async/IndexEntityMessageProcessor.php:82
 Oro\Bundle\SearchBundle\Async\IndexEntityMessageProcessor->process() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Client/DelegateMessageProcessor.php:36
 Oro\Component\MessageQueue\Client\DelegateMessageProcessor->process() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Consumption/QueueConsumer.php:181
 Oro\Component\MessageQueue\Consumption\QueueConsumer->doConsume() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Consumption/QueueConsumer.php:123
 Oro\Component\MessageQueue\Consumption\QueueConsumer->consume() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Client/ConsumeMessagesCommand.php:69
 Oro\Component\MessageQueue\Client\ConsumeMessagesCommand->execute() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Command/Command.php:259
 Symfony\Component\Console\Command\Command->run() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:866
 Symfony\Component\Console\Application->doRunCommand() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:193
 Symfony\Component\Console\Application->doRun() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Bundle/FrameworkBundle/Console/Application.php:92
 Symfony\Bundle\FrameworkBundle\Console\Application->doRun() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:124

For dataaudit:

[error] Consuming interrupted by exception. "An exception occurred while executing 'UPDATE oro_audit SET version=? WHERE id=? AND NOT EXISTS
              (SELECT id FROM
                (SELECT id FROM oro_audit WHERE object_id=? AND object_class=? AND version=?)
              as x)' with params [4, 2402, 390, "Oro\\Bundle\\SalesBundle\\Entity\\Lead", 4]:

SQLSTATE[23505]: Unique violation: 7 ERROR:  duplicate key value violates unique constraint "idx_oro_audit_version"
DETAIL:  Key (object_id, object_class, version)=(390, Oro\Bundle\SalesBundle\Entity\Lead, 4) already exists."
[debug] [RejectMessageOnExceptionDbalExtension] Execution was interrupted and message was rejected. oro.596517363913a3.72749743

  [Doctrine\DBAL\Exception\UniqueConstraintViolationException]                                                         
  An exception occurred while executing 'UPDATE oro_audit SET version=? WHERE id=? AND NOT EXISTS                      
                (SELECT id FROM                                                                                        
                  (SELECT id FROM oro_audit WHERE object_id=? AND object_class=? AND version=?)                        
                as x)' with params [4, 2402, 390, "Oro\\Bundle\\SalesBundle\\Entity\\Lead", 4]:                        
  SQLSTATE[23505]: Unique violation: 7 ERROR:  duplicate key value violates unique constraint "idx_oro_audit_version"  
  DETAIL:  Key (object_id, object_class, version)=(390, Oro\Bundle\SalesBundle\Entity\Lead, 4) already exists.         

Exception trace:
 () at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Driver/AbstractPostgreSQLDriver.php:63
 Doctrine\DBAL\Driver\AbstractPostgreSQLDriver->convertException() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/DBALException.php:128
 Doctrine\DBAL\DBALException::driverExceptionDuringQuery() at /var/www/laboro/application/crm/vendor/doctrine/dbal/lib/Doctrine/DBAL/Connection.php:996
 Doctrine\DBAL\Connection->executeUpdate() at /var/www/laboro/package/platform/src/Oro/Bundle/DataAuditBundle/Service/SetNewAuditVersionService.php:105
 Oro\Bundle\DataAuditBundle\Service\SetNewAuditVersionService->doSetVersion() at /var/www/laboro/package/platform/src/Oro/Bundle/DataAuditBundle/Service/SetNewAuditVersionService.php:63
 Oro\Bundle\DataAuditBundle\Service\SetNewAuditVersionService->setVersion() at /var/www/laboro/package/platform/src/Oro/Bundle/DataAuditBundle/Service/EntityChangesToAuditEntryConverter.php:291
 Oro\Bundle\DataAuditBundle\Service\EntityChangesToAuditEntryConverter->createAuditEntity() at /var/www/laboro/package/platform/src/Oro/Bundle/DataAuditBundle/Service/EntityChangesToAuditEntryConverter.php:102
 Oro\Bundle\DataAuditBundle\Service\EntityChangesToAuditEntryConverter->convert() at /var/www/laboro/package/platform/src/Oro/Bundle/DataAuditBundle/Async/AuditChangedEntitiesProcessor.php:93
 Oro\Bundle\DataAuditBundle\Async\AuditChangedEntitiesProcessor->process() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Client/DelegateMessageProcessor.php:36
 Oro\Component\MessageQueue\Client\DelegateMessageProcessor->process() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Consumption/QueueConsumer.php:181
 Oro\Component\MessageQueue\Consumption\QueueConsumer->doConsume() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Consumption/QueueConsumer.php:123
 Oro\Component\MessageQueue\Consumption\QueueConsumer->consume() at /var/www/laboro/package/platform/src/Oro/Component/MessageQueue/Client/ConsumeMessagesCommand.php:78
 Oro\Component\MessageQueue\Client\ConsumeMessagesCommand->execute() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Command/Command.php:259
 Symfony\Component\Console\Command\Command->run() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:866
 Symfony\Component\Console\Application->doRunCommand() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:193
 Symfony\Component\Console\Application->doRun() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Bundle/FrameworkBundle/Console/Application.php:92
 Symfony\Bundle\FrameworkBundle\Console\Application->doRun() at /var/www/laboro/application/crm/vendor/symfony/symfony/src/Symfony/Component/Console/Application.php:124

Steps to reproduce

0) Environment: orocrm - 2.2, postgressql, dbal for MQ 1) Install orocrm (any versions - master, 2.2, 2.1, 2.0) 2) Change more 10-20x times leadName for entity Oro\Bundle\SalesBundle\Entity\Lead, so that in the message storage there are many messages dataaudit and search (you can use command for it) 3) Run 4 consumers together

Actual result

Two consumers was interrupted by exception

[PDOException]                                                                                                       
  SQLSTATE[23505]: Unique violation: 7 ERROR:  duplicate key value violates unique constraint "idx_oro_audit_version"  
  DETAIL:  Key (object_id, object_class, version)=(390, Oro\Bundle\SalesBundle\Entity\Lead, 4) already exists.

Expected result

There should be no errors

oroinc / platform