From 7b3d803d5923cb4bfe402a0af7173533c700e6a2 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Thu, 3 Jun 2021 19:44:32 +0300
Subject: [PATCH 01/14] 5.0 beta

---
 README.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 5324fc03..01525a15 100644
--- a/README.md
+++ b/README.md
@@ -22,6 +22,7 @@ With this extension, you will be able to sync the following data from Akeneo to
 |    3.1    |  EOL   |     3.1     | 2.3, 3.2, 4.0* |       |
 |    4.1    |  2021  |     4.1     | 2.3, 3.2, 4.0* | [![Build Status](https://travis-ci.org/oroinc/OroAkeneoBundle.svg?branch=4.1)](https://travis-ci.org/oroinc/OroAkeneoBundle) |
 |    4.2    |  2022  |     4.2     | 3.2, 4.0, 5.0  | [![Build Status](https://travis-ci.org/oroinc/OroAkeneoBundle.svg?branch=4.2)](https://travis-ci.org/oroinc/OroAkeneoBundle) |
+|    5.0    |  WIP   |     5.0     | 4.0, 5.0       |       |
 
 ** Akeneo supported using older client versions, new features are not available.**
 
@@ -30,7 +31,7 @@ With this extension, you will be able to sync the following data from Akeneo to
 1. Add composer package
 
 ```
-# Akeneo 3.2 and 4.0
+# Akeneo 4.0
 composer require "oro/commerce-akeneo:4.2.*" "akeneo/api-php-client-ee:5.*"
 
 # Akeneo 5.0

From 2c5dba6e43f5aa32fe5289d949a0768d88d96644 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Thu, 3 Jun 2021 19:45:45 +0300
Subject: [PATCH 02/14] PHP 8.0

---
 .travis.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.yml b/.travis.yml
index 967ac9e9..40ecb0f3 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,7 +1,7 @@
 language: php
 
 php:
-    - '7.4'
+    - '8.0'
 
 install:
     - php -r "copy('https://cs.symfony.com/download/php-cs-fixer-v2.phar', 'php-cs-fixer.phar');"

From 28e392f04cb366ab7c05a95a07158c5db132ec6c Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Mon, 16 Aug 2021 15:27:30 +0300
Subject: [PATCH 03/14] Report SSL connection errors

---
 Controller/ValidateConnectionController.php | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/Controller/ValidateConnectionController.php b/Controller/ValidateConnectionController.php
index ecca092b..219f3c5e 100644
--- a/Controller/ValidateConnectionController.php
+++ b/Controller/ValidateConnectionController.php
@@ -7,6 +7,7 @@
 use Oro\Bundle\IntegrationBundle\Entity\Channel;
 use Oro\Bundle\IntegrationBundle\Form\Type\ChannelType;
 use Oro\Bundle\SecurityBundle\Annotation\Acl;
+use Psr\Http\Client\ClientExceptionInterface;
 use Sensio\Bundle\FrameworkExtraBundle\Configuration\ParamConverter;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\HttpFoundation\JsonResponse;
@@ -78,7 +79,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
                     $akeneoCurrencies = $transport->getMergedCurrencies();
                     $akeneoLocales = $transport->getLocales();
             }
-        } catch (ExceptionInterface $e) {
+        } catch (ClientExceptionInterface | ExceptionInterface $e) {
             $success = false;
             $message = $e->getMessage();
         } catch (\Exception $e) {

From 0e70f4035ddf7e2009e8495601f8655ebf104fba Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Tue, 31 Aug 2021 21:10:05 +0300
Subject: [PATCH 04/14] AKM-29: Memory leak in api step - flush removed from
 the core causes
 \Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer::flushBuffer
 skip - causes \Oro\Bundle\AkeneoBundle\ImportExport\Writer\AsyncWriter to
 keep messages and json data in memory -
 \Akeneo\Pim\ApiClient\Pagination\Page::getPage do not free up memory because
 of json data in use

---
 ImportExport/Writer/AsyncWriter.php | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/ImportExport/Writer/AsyncWriter.php b/ImportExport/Writer/AsyncWriter.php
index 1ff2025b..417d9047 100644
--- a/ImportExport/Writer/AsyncWriter.php
+++ b/ImportExport/Writer/AsyncWriter.php
@@ -8,6 +8,7 @@
 use Doctrine\Common\Cache\CacheProvider;
 use Oro\Bundle\AkeneoBundle\Async\Topics;
 use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
+use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
 use Oro\Component\MessageQueue\Client\Message;
 use Oro\Component\MessageQueue\Client\MessagePriority;
 use Oro\Component\MessageQueue\Client\MessageProducerInterface;
@@ -104,6 +105,11 @@ function (JobRunner $jobRunner, Job $child) use ($items, $channelId) {
                         )
                     );
 
+                    if ($this->messageProducer instanceof BufferedMessageProducer
+                        && $this->messageProducer->isBufferingEnabled()) {
+                        $this->messageProducer->flushBuffer();
+                    }
+
                     return true;
                 }
             );

From 549b3e7402e9e56dd57e91374addbd9cc2615753 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Tue, 5 Oct 2021 21:23:14 +0300
Subject: [PATCH 05/14] BC compat with master

---
 Controller/ValidateConnectionController.php   | 47 +++++++++++++------
 DependencyInjection/Configuration.php         | 17 +------
 DependencyInjection/OroAkeneoExtension.php    |  1 +
 EventListener/AttributesDatagridListener.php  |  2 +-
 .../DeletedAttributeRelationListener.php      | 19 +++++++-
 .../ImportExportTagsSubscriberDecorator.php   | 14 +++++-
 Form/Extension/ChannelTypeExtension.php       |  4 +-
 Form/Extension/ProductTypeExtension.php       |  4 +-
 Form/Type/AkeneoSettingsType.php              |  2 +-
 .../DataConverter/AttributeDataConverter.php  |  4 +-
 .../DataConverter/BrandDataConverter.php      |  3 +-
 .../DataConverter/ProductDataConverter.php    |  3 +-
 .../Processor/AttributeImportProcessor.php    |  2 +-
 .../Processor/ProductVariantProcessor.php     |  6 +--
 .../Normalizer/AkeneoNormalizerWrapper.php    |  8 ++--
 .../Normalizer/EntityFieldNormalizer.php      |  2 +-
 .../Normalizer/FileItemNormalizer.php         |  8 ++--
 ImportExport/Step/ItemStep.php                |  2 +-
 ImportExport/Step/StepExecutor.php            |  4 +-
 .../Strategy/ImportStrategyHelper.php         |  9 ++--
 .../Strategy/StrategyRelationsTrait.php       |  6 ++-
 ImportExport/Writer/AsyncWriter.php           |  9 ++--
 ImportExport/Writer/AttributeWriter.php       |  4 +-
 ImportExport/Writer/CumulativeWriter.php      |  6 +--
 Integration/AkeneoTransport.php               |  4 +-
 Integration/Connector/AttributeConnector.php  |  2 +-
 .../Connector/AttributeFamilyConnector.php    |  2 +-
 Integration/Connector/BrandConnector.php      |  2 +-
 Integration/Connector/CategoryConnector.php   |  2 +-
 Integration/Connector/ProductConnector.php    |  2 +-
 Job/Context/SimpleContextAggregator.php       |  2 +-
 Migrations/Schema/v1_8/OroAkeneoMigration.php |  3 +-
 Resources/config/batch_jobs.yml               | 14 +++---
 Resources/config/controllers.yml              | 13 +++++
 Resources/config/oro/placeholders.yml         |  2 +-
 Resources/config/oro/twig.yml                 |  2 +-
 Resources/config/services.yml                 | 18 ++++---
 .../views/Button/schema_update.html.twig      |  2 +-
 .../Datagrid/attributeFamilies.html.twig      |  2 +-
 Resources/views/Form/fields.html.twig         |  2 +-
 composer.json                                 |  2 +-
 41 files changed, 160 insertions(+), 102 deletions(-)
 create mode 100644 Resources/config/controllers.yml

diff --git a/Controller/ValidateConnectionController.php b/Controller/ValidateConnectionController.php
index 219f3c5e..7cb5c322 100644
--- a/Controller/ValidateConnectionController.php
+++ b/Controller/ValidateConnectionController.php
@@ -4,6 +4,8 @@
 
 use Akeneo\Pim\ApiClient\Exception\ExceptionInterface;
 use Oro\Bundle\AkeneoBundle\Entity\AkeneoSettings;
+use Oro\Bundle\AkeneoBundle\Integration\AkeneoTransportInterface;
+use Oro\Bundle\CurrencyBundle\Provider\CurrencyProviderInterface;
 use Oro\Bundle\IntegrationBundle\Entity\Channel;
 use Oro\Bundle\IntegrationBundle\Form\Type\ChannelType;
 use Oro\Bundle\SecurityBundle\Annotation\Acl;
@@ -13,12 +15,32 @@
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Request;
 use Symfony\Component\Routing\Annotation\Route;
+use Symfony\Contracts\Translation\TranslatorInterface;
 
 class ValidateConnectionController extends AbstractController
 {
     const CONNECTION_SUCCESSFUL_MESSAGE = 'oro.akeneo.connection.successfull';
     const CONNECTION_ERROR_MESSAGE = 'oro.akeneo.connection.error';
 
+    /** @var CurrencyProviderInterface */
+    private $currencyProvider;
+
+    /** @var TranslatorInterface */
+    private $translator;
+
+    /** @var AkeneoTransportInterface */
+    private $akeneoTransport;
+
+    public function __construct(
+        CurrencyProviderInterface $currencyProvider,
+        TranslatorInterface $translator,
+        AkeneoTransportInterface $akeneoTransport
+    ) {
+        $this->currencyProvider = $currencyProvider;
+        $this->translator = $translator;
+        $this->akeneoTransport = $akeneoTransport;
+    }
+
     /**
      * @Route(path="/validate-akeneo-connection/{channelId}/", name="oro_akeneo_validate_connection", methods={"POST"})
      * @ParamConverter("channel", class="OroIntegrationBundle:Channel", options={"id"="channelId"})
@@ -53,38 +75,35 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
             $akeneoSettings->setPassword($akeneoSettingsEntity->getPassword());
         }
 
-        $currencyConfig = $this->container->get('oro_currency.config.currency');
-
         $akeneoChannelNames = [];
         $akeneoCurrencies = [];
         $akeneoLocales = [];
 
         try {
-            $transport = $this->get('oro_akeneo.integration.transport');
-            $transport->init($akeneoSettings, false);
+            $this->akeneoTransport->init($akeneoSettings, false);
             $success = true;
-            $message = self::CONNECTION_SUCCESSFUL_MESSAGE;
+            $message = $this->translator->trans(self::CONNECTION_SUCCESSFUL_MESSAGE);
             switch ($request->get('synctype', 'all')) {
                 case 'channels':
-                    $akeneoChannelNames = $transport->getChannels();
+                    $akeneoChannelNames = $this->akeneoTransport->getChannels();
                     break;
                 case 'currencies':
-                    $akeneoCurrencies = $transport->getMergedCurrencies();
+                    $akeneoCurrencies = $this->akeneoTransport->getMergedCurrencies();
                     break;
                 case 'locales':
-                    $akeneoLocales = $transport->getLocales();
+                    $akeneoLocales = $this->akeneoTransport->getLocales();
                     break;
                 default:
-                    $akeneoChannelNames = $transport->getChannels();
-                    $akeneoCurrencies = $transport->getMergedCurrencies();
-                    $akeneoLocales = $transport->getLocales();
+                    $akeneoChannelNames = $this->akeneoTransport->getChannels();
+                    $akeneoCurrencies = $this->akeneoTransport->getMergedCurrencies();
+                    $akeneoLocales = $this->akeneoTransport->getLocales();
             }
         } catch (ClientExceptionInterface | ExceptionInterface $e) {
             $success = false;
             $message = $e->getMessage();
         } catch (\Exception $e) {
             $success = false;
-            $message = self::CONNECTION_ERROR_MESSAGE;
+            $message = $this->translator->trans(self::CONNECTION_ERROR_MESSAGE);
         }
 
         return new JsonResponse(
@@ -93,8 +112,8 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
                 'akeneoCurrencies' => $akeneoCurrencies,
                 'akeneoLocales' => $akeneoLocales,
                 'success' => $success,
-                'message' => $this->get('translator')->trans($message),
-                'currencyList' => $currencyConfig->getCurrencies(),
+                'message' => $message,
+                'currencyList' => $this->currencyProvider->getCurrencies(),
             ]
         );
     }
diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
index 6f7858d3..ce3a808b 100644
--- a/DependencyInjection/Configuration.php
+++ b/DependencyInjection/Configuration.php
@@ -5,25 +5,10 @@
 use Symfony\Component\Config\Definition\Builder\TreeBuilder;
 use Symfony\Component\Config\Definition\ConfigurationInterface;
 
-/**
- * This is the class that validates and merges configuration from your app/config files.
- *
- * To learn more see {@link http://symfony.com/doc/current/cookbook/bundles/configuration.html}
- */
 class Configuration implements ConfigurationInterface
 {
-    /**
-     * {@inheritdoc}
-     */
     public function getConfigTreeBuilder()
     {
-        $treeBuilder = new TreeBuilder();
-        $rootNode = $treeBuilder->root('oro_akeneo');
-
-        // Here you should define the parameters that are allowed to
-        // configure your bundle. See the documentation linked above for
-        // more information on that topic.
-
-        return $treeBuilder;
+        return new TreeBuilder('oro_akeneo');
     }
 }
diff --git a/DependencyInjection/OroAkeneoExtension.php b/DependencyInjection/OroAkeneoExtension.php
index 64d5af6f..fcf59856 100644
--- a/DependencyInjection/OroAkeneoExtension.php
+++ b/DependencyInjection/OroAkeneoExtension.php
@@ -23,6 +23,7 @@ public function load(array $configs, ContainerBuilder $container)
     {
         $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
         $loader->load('integration.yml');
+        $loader->load('controllers.yml');
         $loader->load('importexport.yml');
         $loader->load('services.yml');
         $loader->load('form_types.yml');
diff --git a/EventListener/AttributesDatagridListener.php b/EventListener/AttributesDatagridListener.php
index 894e6b4d..7367eabb 100644
--- a/EventListener/AttributesDatagridListener.php
+++ b/EventListener/AttributesDatagridListener.php
@@ -23,7 +23,7 @@ public function onBuildBefore(BuildBefore $event)
     {
         $event->getConfig()->offsetSetByPath(
             '[columns][attributeFamilies][template]',
-            'OroAkeneoBundle:Datagrid:attributeFamilies.html.twig'
+            '@OroAkeneo/Datagrid/attributeFamilies.html.twig'
         );
     }
 
diff --git a/EventListener/DeletedAttributeRelationListener.php b/EventListener/DeletedAttributeRelationListener.php
index 126fca56..553f262b 100644
--- a/EventListener/DeletedAttributeRelationListener.php
+++ b/EventListener/DeletedAttributeRelationListener.php
@@ -2,18 +2,33 @@
 
 namespace Oro\Bundle\AkeneoBundle\EventListener;
 
-use Doctrine\Common\Inflector\Inflector;
+use Doctrine\Inflector\Inflector;
 use Doctrine\ORM\Event\OnFlushEventArgs;
 use Oro\Bundle\EntityConfigBundle\Attribute\Entity\AttributeGroupRelation;
 use Oro\Bundle\EntityConfigBundle\EventListener\DeletedAttributeRelationListener as BaseListener;
+use Oro\Bundle\EntityConfigBundle\Provider\DeletedAttributeProviderInterface;
 use Oro\Component\MessageQueue\Client\Message;
 use Oro\Component\MessageQueue\Client\MessagePriority;
+use Oro\Component\MessageQueue\Client\MessageProducerInterface;
 
 class DeletedAttributeRelationListener extends BaseListener
 {
     /** @var array */
     protected $deletedAttributesNames = [];
 
+    /** @var Inflector */
+    private $inflector;
+
+    public function __construct(
+        MessageProducerInterface $messageProducer,
+        DeletedAttributeProviderInterface $deletedAttributeProvider,
+        Inflector $inflector
+    ) {
+        parent::__construct($messageProducer, $deletedAttributeProvider, $inflector);
+
+        $this->inflector = $inflector;
+    }
+
     public function onFlush(OnFlushEventArgs $eventArgs)
     {
         $uow = $eventArgs->getEntityManager()->getUnitOfWork();
@@ -32,7 +47,7 @@ public function onFlush(OnFlushEventArgs $eventArgs)
         foreach ($this->deletedAttributes as $attributeFamilyId => $attributeIds) {
             $attributes = $this->deletedAttributeProvider->getAttributesByIds($attributeIds);
             foreach ($attributes as &$attribute) {
-                $attribute = Inflector::camelize($attribute->getFieldName());
+                $attribute = $this->inflector->camelize($attribute->getFieldName());
             }
 
             $this->deletedAttributesNames[$attributeFamilyId] = array_merge(
diff --git a/EventListener/ImportExportTagsSubscriberDecorator.php b/EventListener/ImportExportTagsSubscriberDecorator.php
index 9b42c3ea..63c25167 100644
--- a/EventListener/ImportExportTagsSubscriberDecorator.php
+++ b/EventListener/ImportExportTagsSubscriberDecorator.php
@@ -11,12 +11,17 @@
 use Oro\Bundle\ImportExportBundle\Event\NormalizeEntityEvent;
 use Oro\Bundle\ImportExportBundle\Event\StrategyEvent;
 use Oro\Bundle\TagBundle\EventListener\ImportExportTagsSubscriber;
+use Oro\Bundle\TagBundle\Manager\TagImportManager;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
+use Symfony\Contracts\Service\ServiceSubscriberInterface;
 
 /**
  * Tags lazy processing
  */
-class ImportExportTagsSubscriberDecorator implements AdditionalOptionalListenerInterface, EventSubscriberInterface
+class ImportExportTagsSubscriberDecorator implements
+    AdditionalOptionalListenerInterface,
+    EventSubscriberInterface,
+    ServiceSubscriberInterface
 {
     use AdditionalOptionalListenerTrait;
 
@@ -116,4 +121,11 @@ public function addTagsIntoFixtures(LoadTemplateFixturesEvent $event): void
 
         $this->innerSubscriber->addTagsIntoFixtures($event);
     }
+
+    public static function getSubscribedServices()
+    {
+        return [
+            'oro_tag.tag_import.manager' => TagImportManager::class,
+        ];
+    }
 }
diff --git a/Form/Extension/ChannelTypeExtension.php b/Form/Extension/ChannelTypeExtension.php
index 6b7cdecc..86ec2363 100644
--- a/Form/Extension/ChannelTypeExtension.php
+++ b/Form/Extension/ChannelTypeExtension.php
@@ -24,9 +24,9 @@ class ChannelTypeExtension extends AbstractTypeExtension
     /**
      * {@inheritdoc}
      */
-    public function getExtendedType()
+    public static function getExtendedTypes(): iterable
     {
-        return ChannelType::class;
+        return [ChannelType::class];
     }
 
     /**
diff --git a/Form/Extension/ProductTypeExtension.php b/Form/Extension/ProductTypeExtension.php
index 03cf1fea..eea94770 100644
--- a/Form/Extension/ProductTypeExtension.php
+++ b/Form/Extension/ProductTypeExtension.php
@@ -42,9 +42,9 @@ public function __construct(ConfigManager $configManager, FieldHelper $fieldHelp
     /**
      * {@inheritdoc}
      */
-    public function getExtendedType()
+    public static function getExtendedTypes(): iterable
     {
-        return ProductType::class;
+        return [ProductType::class];
     }
 
     /**
diff --git a/Form/Type/AkeneoSettingsType.php b/Form/Type/AkeneoSettingsType.php
index ae296418..77f8b508 100644
--- a/Form/Type/AkeneoSettingsType.php
+++ b/Form/Type/AkeneoSettingsType.php
@@ -26,11 +26,11 @@
 use Symfony\Component\Form\FormEvents;
 use Symfony\Component\OptionsResolver\Exception\AccessException;
 use Symfony\Component\OptionsResolver\OptionsResolver;
-use Symfony\Component\Translation\TranslatorInterface;
 use Symfony\Component\Validator\Constraints\NotBlank;
 use Symfony\Component\Validator\Exception\ConstraintDefinitionException;
 use Symfony\Component\Validator\Exception\InvalidOptionsException;
 use Symfony\Component\Validator\Exception\MissingOptionsException;
+use Symfony\Contracts\Translation\TranslatorInterface;
 
 class AkeneoSettingsType extends AbstractType implements LoggerAwareInterface
 {
diff --git a/ImportExport/DataConverter/AttributeDataConverter.php b/ImportExport/DataConverter/AttributeDataConverter.php
index 21f66e86..6b18c091 100644
--- a/ImportExport/DataConverter/AttributeDataConverter.php
+++ b/ImportExport/DataConverter/AttributeDataConverter.php
@@ -2,14 +2,14 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\DataConverter;
 
-use Akeneo\Bundle\BatchBundle\Item\InvalidItemException;
 use Oro\Bundle\AkeneoBundle\ImportExport\AkeneoIntegrationTrait;
 use Oro\Bundle\AkeneoBundle\Tools\AttributeTypeConverter;
 use Oro\Bundle\AkeneoBundle\Tools\FieldConfigModelFieldNameGenerator;
 use Oro\Bundle\AkeneoBundle\Tools\Generator;
+use Oro\Bundle\BatchBundle\Exception\InvalidItemException;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
 use Oro\Bundle\EntityConfigBundle\ImportExport\DataConverter\EntityFieldDataConverter;
-use Symfony\Component\Translation\TranslatorInterface;
+use Symfony\Contracts\Translation\TranslatorInterface;
 
 /**
  * Converts data to import format.
diff --git a/ImportExport/DataConverter/BrandDataConverter.php b/ImportExport/DataConverter/BrandDataConverter.php
index 8cc18730..95cd3e65 100644
--- a/ImportExport/DataConverter/BrandDataConverter.php
+++ b/ImportExport/DataConverter/BrandDataConverter.php
@@ -8,6 +8,7 @@
 use Oro\Bundle\ConfigBundle\Config\ConfigManager;
 use Oro\Bundle\EntityBundle\Helper\FieldHelper;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\EntityBundle\Provider\EntityFieldProvider;
 use Oro\Bundle\EntityConfigBundle\Config\ConfigManager as EntityConfigManager;
 use Oro\Bundle\EntityExtendBundle\Extend\RelationType;
 use Oro\Bundle\ImportExportBundle\Context\ContextAwareInterface;
@@ -64,7 +65,7 @@ public function convertToExportFormat(array $exportedRecord, $skipNullValues = t
     public function convertToImportFormat(array $importedRecord, $skipNullValues = true)
     {
         $record = [];
-        $fields = $this->fieldHelper->getFields(Brand::class, true);
+        $fields = $this->fieldHelper->getEntityFields(Brand::class, EntityFieldProvider::OPTION_WITH_RELATIONS);
         $fieldsByName = [];
         foreach ($fields as $field) {
             $fieldsByName[$field['name']] = $field;
diff --git a/ImportExport/DataConverter/ProductDataConverter.php b/ImportExport/DataConverter/ProductDataConverter.php
index 867ecd6d..a16a50ce 100644
--- a/ImportExport/DataConverter/ProductDataConverter.php
+++ b/ImportExport/DataConverter/ProductDataConverter.php
@@ -7,6 +7,7 @@
 use Oro\Bundle\AkeneoBundle\Tools\AttributeFamilyCodeGenerator;
 use Oro\Bundle\AkeneoBundle\Tools\Generator;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\EntityBundle\Provider\EntityFieldProvider;
 use Oro\Bundle\EntityConfigBundle\Config\ConfigManager;
 use Oro\Bundle\EntityExtendBundle\Extend\RelationType;
 use Oro\Bundle\ImportExportBundle\Context\ContextAwareInterface;
@@ -321,7 +322,7 @@ private function prepareFieldMapping()
             return;
         }
 
-        $fields = $this->fieldHelper->getFields(Product::class, true);
+        $fields = $this->fieldHelper->getEntityFields(Product::class, EntityFieldProvider::OPTION_WITH_RELATIONS);
         $importExportProvider = $this->entityConfigManager->getProvider('importexport');
 
         foreach ($fields as $field) {
diff --git a/ImportExport/Processor/AttributeImportProcessor.php b/ImportExport/Processor/AttributeImportProcessor.php
index f794dd62..f04f63a2 100644
--- a/ImportExport/Processor/AttributeImportProcessor.php
+++ b/ImportExport/Processor/AttributeImportProcessor.php
@@ -136,7 +136,7 @@ public function setConfigManager($configManager)
         $this->configManager = $configManager;
     }
 
-    public function setImportExportContext(ContextInterface $context)
+    public function setImportExportContext(ContextInterface $context): void
     {
         $context->setValue('entity_id', $this->configManager->getConfigModelId($this->entityConfigModelClassName));
 
diff --git a/ImportExport/Processor/ProductVariantProcessor.php b/ImportExport/Processor/ProductVariantProcessor.php
index cb3b173d..5e564705 100644
--- a/ImportExport/Processor/ProductVariantProcessor.php
+++ b/ImportExport/Processor/ProductVariantProcessor.php
@@ -2,16 +2,16 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Processor;
 
-use Akeneo\Bundle\BatchBundle\Entity\StepExecution;
-use Akeneo\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Doctrine\Persistence\ManagerRegistry;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Oro\Bundle\ImportExportBundle\Context\ContextRegistry;
 use Oro\Bundle\ImportExportBundle\Processor\ProcessorInterface;
 use Oro\Bundle\ImportExportBundle\Strategy\Import\ImportStrategyHelper;
 use Oro\Bundle\ProductBundle\Entity\Product;
 use Oro\Bundle\ProductBundle\Entity\ProductVariantLink;
 use Oro\Bundle\ProductBundle\Entity\Repository\ProductRepository;
-use Symfony\Component\Translation\TranslatorInterface;
+use Symfony\Contracts\Translation\TranslatorInterface;
 
 class ProductVariantProcessor implements ProcessorInterface, StepExecutionAwareInterface
 {
diff --git a/ImportExport/Serializer/Normalizer/AkeneoNormalizerWrapper.php b/ImportExport/Serializer/Normalizer/AkeneoNormalizerWrapper.php
index 34e09b1c..f3446df9 100644
--- a/ImportExport/Serializer/Normalizer/AkeneoNormalizerWrapper.php
+++ b/ImportExport/Serializer/Normalizer/AkeneoNormalizerWrapper.php
@@ -3,14 +3,14 @@
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Serializer\Normalizer;
 
 use Oro\Bundle\AkeneoBundle\Integration\AkeneoChannel;
-use Oro\Bundle\ImportExportBundle\Serializer\Normalizer\DenormalizerInterface;
+use Symfony\Component\Serializer\Normalizer\ContextAwareDenormalizerInterface;
 
-class AkeneoNormalizerWrapper implements DenormalizerInterface
+class AkeneoNormalizerWrapper implements ContextAwareDenormalizerInterface
 {
-    /** @var DenormalizerInterface */
+    /** @var ContextAwareDenormalizerInterface */
     private $fileNormalizer;
 
-    public function __construct(DenormalizerInterface $fileNormalizer)
+    public function __construct(ContextAwareDenormalizerInterface $fileNormalizer)
     {
         $this->fileNormalizer = $fileNormalizer;
     }
diff --git a/ImportExport/Serializer/Normalizer/EntityFieldNormalizer.php b/ImportExport/Serializer/Normalizer/EntityFieldNormalizer.php
index ea3b0f6b..8e6ff281 100644
--- a/ImportExport/Serializer/Normalizer/EntityFieldNormalizer.php
+++ b/ImportExport/Serializer/Normalizer/EntityFieldNormalizer.php
@@ -10,7 +10,7 @@ class EntityFieldNormalizer extends BaseEntityFieldNormalizer
     /**
      * {@inheritdoc}
      */
-    public function supportsDenormalization($data, $type, $format = null, array $context = [])
+    public function supportsDenormalization($data, $type, $format = null, array $context = []): bool
     {
         return is_array($data)
             && is_a($type, 'Oro\Bundle\EntityConfigBundle\Entity\FieldConfigModel', true)
diff --git a/ImportExport/Serializer/Normalizer/FileItemNormalizer.php b/ImportExport/Serializer/Normalizer/FileItemNormalizer.php
index f482730c..651978f6 100644
--- a/ImportExport/Serializer/Normalizer/FileItemNormalizer.php
+++ b/ImportExport/Serializer/Normalizer/FileItemNormalizer.php
@@ -4,14 +4,14 @@
 
 use Oro\Bundle\AkeneoBundle\Integration\AkeneoChannel;
 use Oro\Bundle\AttachmentBundle\Entity\FileItem;
-use Oro\Bundle\ImportExportBundle\Serializer\Normalizer\DenormalizerInterface;
+use Symfony\Component\Serializer\Normalizer\ContextAwareDenormalizerInterface;
 
-class FileItemNormalizer implements DenormalizerInterface
+class FileItemNormalizer implements ContextAwareDenormalizerInterface
 {
-    /** @var DenormalizerInterface */
+    /** @var ContextAwareDenormalizerInterface */
     private $fileNormalizer;
 
-    public function __construct(DenormalizerInterface $fileNormalizer)
+    public function __construct(ContextAwareDenormalizerInterface $fileNormalizer)
     {
         $this->fileNormalizer = $fileNormalizer;
     }
diff --git a/ImportExport/Step/ItemStep.php b/ImportExport/Step/ItemStep.php
index 638b108d..af787c3c 100644
--- a/ImportExport/Step/ItemStep.php
+++ b/ImportExport/Step/ItemStep.php
@@ -2,7 +2,7 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Step;
 
-use Akeneo\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
 use Oro\Bundle\BatchBundle\Step\ItemStep as BaseItemStep;
 
 class ItemStep extends BaseItemStep
diff --git a/ImportExport/Step/StepExecutor.php b/ImportExport/Step/StepExecutor.php
index 4c104024..90674334 100644
--- a/ImportExport/Step/StepExecutor.php
+++ b/ImportExport/Step/StepExecutor.php
@@ -2,13 +2,13 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Step;
 
-use Akeneo\Bundle\BatchBundle\Item\InvalidItemException;
+use Oro\Bundle\BatchBundle\Exception\InvalidItemException;
 use Oro\Bundle\BatchBundle\Step\StepExecutionWarningHandlerInterface;
 use Oro\Bundle\BatchBundle\Step\StepExecutor as BaseStepExecutor;
 
 class StepExecutor extends BaseStepExecutor
 {
-    public function execute(StepExecutionWarningHandlerInterface $warningHandler = null)
+    public function execute(StepExecutionWarningHandlerInterface $warningHandler = null): void
     {
         try {
             $stopExecution = false;
diff --git a/ImportExport/Strategy/ImportStrategyHelper.php b/ImportExport/Strategy/ImportStrategyHelper.php
index 06b1d251..7cbeae8b 100644
--- a/ImportExport/Strategy/ImportStrategyHelper.php
+++ b/ImportExport/Strategy/ImportStrategyHelper.php
@@ -6,6 +6,7 @@
 use Doctrine\Common\Util\ClassUtils;
 use Oro\Bundle\AkeneoBundle\Integration\AkeneoChannel;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\EntityBundle\Provider\EntityFieldProvider;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 use Oro\Bundle\ImportExportBundle\Exception\InvalidArgumentException;
 use Oro\Bundle\ImportExportBundle\Strategy\Import\ImportStrategyHelper as BaseImportStrategyHelper;
@@ -101,9 +102,9 @@ protected function getEntityPropertiesByClassName($entityClassName)
          * that mustn't be changed by import/export
          */
         if ($this->extendConfigProvider->hasConfig($entityClassName)) {
-            $properties = $this->fieldHelper->getFields(
+            $properties = $this->fieldHelper->getEntityFields(
                 $entityClassName,
-                true
+                EntityFieldProvider::OPTION_WITH_RELATIONS
             );
 
             return array_column($properties, 'name');
@@ -122,7 +123,9 @@ protected function getEntityPropertiesByClassName($entityClassName)
     public function addValidationErrors(array $validationErrors, ContextInterface $context, $errorPrefix = null)
     {
         if (AkeneoChannel::TYPE !== $context->getOption('channelType')) {
-            return parent::addValidationErrors($validationErrors, $context, $errorPrefix);
+            parent::addValidationErrors($validationErrors, $context, $errorPrefix);
+
+            return;
         }
 
         foreach ($validationErrors as $validationError) {
diff --git a/ImportExport/Strategy/StrategyRelationsTrait.php b/ImportExport/Strategy/StrategyRelationsTrait.php
index eb0e427d..d96be265 100644
--- a/ImportExport/Strategy/StrategyRelationsTrait.php
+++ b/ImportExport/Strategy/StrategyRelationsTrait.php
@@ -3,6 +3,7 @@
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Strategy;
 
 use Doctrine\Common\Collections\Collection;
+use Oro\Bundle\EntityBundle\Provider\EntityFieldProvider;
 
 /**
  * @property $doctrineHelper
@@ -28,7 +29,10 @@ trait StrategyRelationsTrait
     protected function updateRelations($entity, array $itemData = null)
     {
         $entityName = $this->doctrineHelper->getEntityClass($entity);
-        $fields = $this->fieldHelper->getFields($entityName, true);
+        $fields = $this->fieldHelper->getEntityFields(
+            $entityName,
+            EntityFieldProvider::OPTION_WITH_RELATIONS
+        );
 
         foreach ($fields as $field) {
             if ($this->fieldHelper->isRelation($field)) {
diff --git a/ImportExport/Writer/AsyncWriter.php b/ImportExport/Writer/AsyncWriter.php
index 417d9047..e223401d 100644
--- a/ImportExport/Writer/AsyncWriter.php
+++ b/ImportExport/Writer/AsyncWriter.php
@@ -2,12 +2,12 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer;
 
-use Akeneo\Bundle\BatchBundle\Entity\StepExecution;
-use Akeneo\Bundle\BatchBundle\Item\ItemWriterInterface;
-use Akeneo\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Doctrine\Common\Cache\CacheProvider;
 use Oro\Bundle\AkeneoBundle\Async\Topics;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
 use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
+use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
 use Oro\Component\MessageQueue\Client\Message;
 use Oro\Component\MessageQueue\Client\MessagePriority;
@@ -105,8 +105,7 @@ function (JobRunner $jobRunner, Job $child) use ($items, $channelId) {
                         )
                     );
 
-                    if ($this->messageProducer instanceof BufferedMessageProducer
-                        && $this->messageProducer->isBufferingEnabled()) {
+                    if ($this->messageProducer instanceof BufferedMessageProducer && $this->messageProducer->isBufferingEnabled()) {
                         $this->messageProducer->flushBuffer();
                     }
 
diff --git a/ImportExport/Writer/AttributeWriter.php b/ImportExport/Writer/AttributeWriter.php
index 701e5559..c78ca4f0 100644
--- a/ImportExport/Writer/AttributeWriter.php
+++ b/ImportExport/Writer/AttributeWriter.php
@@ -2,11 +2,11 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer;
 
-use Akeneo\Bundle\BatchBundle\Entity\StepExecution;
-use Akeneo\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Doctrine\Common\Cache\CacheProvider;
 use Oro\Bundle\AkeneoBundle\Config\ChangesAwareInterface;
 use Oro\Bundle\AkeneoBundle\Tools\EnumSynchronizer;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Oro\Bundle\EntityBundle\EntityConfig\DatagridScope;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
 use Oro\Bundle\EntityConfigBundle\Attribute\AttributeTypeRegistry;
diff --git a/ImportExport/Writer/CumulativeWriter.php b/ImportExport/Writer/CumulativeWriter.php
index e040929c..55924b2f 100644
--- a/ImportExport/Writer/CumulativeWriter.php
+++ b/ImportExport/Writer/CumulativeWriter.php
@@ -2,15 +2,15 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer;
 
-use Akeneo\Bundle\BatchBundle\Entity\StepExecution;
-use Akeneo\Bundle\BatchBundle\Item\ItemWriterInterface;
-use Akeneo\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Doctrine\Common\Util\ClassUtils;
 use Doctrine\ORM\EntityManager;
 use Doctrine\ORM\UnitOfWork;
 use Doctrine\Persistence\ManagerRegistry;
 use Oro\Bundle\AkeneoBundle\EventListener\AdditionalOptionalListenerManager;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
 use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
+use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
 use Oro\Bundle\BatchBundle\Step\StepExecutionRestoreInterface;
 use Oro\Bundle\PlatformBundle\Manager\OptionalListenerManager;
 
diff --git a/Integration/AkeneoTransport.php b/Integration/AkeneoTransport.php
index 14a9fb4d..a6f87e31 100644
--- a/Integration/AkeneoTransport.php
+++ b/Integration/AkeneoTransport.php
@@ -17,7 +17,7 @@
 use Oro\Bundle\IntegrationBundle\Entity\Transport;
 use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
-use Symfony\Component\Intl\Intl;
+use Symfony\Component\Intl\Languages;
 
 class AkeneoTransport implements AkeneoTransportInterface
 {
@@ -131,7 +131,7 @@ public function getLocales()
                 continue;
             }
 
-            $localeName = Intl::getLocaleBundle()->getLocaleName($locale['code']);
+            $localeName = Languages::getName($locale['code']);
             $locales[$localeName ?: $locale['code']] = $locale['code'];
         }
 
diff --git a/Integration/Connector/AttributeConnector.php b/Integration/Connector/AttributeConnector.php
index 0ad33789..f061b8a3 100644
--- a/Integration/Connector/AttributeConnector.php
+++ b/Integration/Connector/AttributeConnector.php
@@ -17,7 +17,7 @@ class AttributeConnector extends AbstractConnector
     /**
      * {@inheritdoc}
      */
-    public function getLabel()
+    public function getLabel(): string
     {
         return 'oro.akeneo.connector.attribute.label';
     }
diff --git a/Integration/Connector/AttributeFamilyConnector.php b/Integration/Connector/AttributeFamilyConnector.php
index 6c1ce2b9..325a8f9a 100644
--- a/Integration/Connector/AttributeFamilyConnector.php
+++ b/Integration/Connector/AttributeFamilyConnector.php
@@ -16,7 +16,7 @@ class AttributeFamilyConnector extends AbstractConnector
     /**
      * {@inheritdoc}
      */
-    public function getLabel()
+    public function getLabel(): string
     {
         return 'oro.akeneo.connector.attribute_family.label';
     }
diff --git a/Integration/Connector/BrandConnector.php b/Integration/Connector/BrandConnector.php
index c4f29c57..ecbddf58 100644
--- a/Integration/Connector/BrandConnector.php
+++ b/Integration/Connector/BrandConnector.php
@@ -11,7 +11,7 @@
  */
 class BrandConnector extends AbstractConnector
 {
-    public function getLabel()
+    public function getLabel(): string
     {
         return 'oro.akeneo.connector.brand.label';
     }
diff --git a/Integration/Connector/CategoryConnector.php b/Integration/Connector/CategoryConnector.php
index 96626a89..9caf9d93 100644
--- a/Integration/Connector/CategoryConnector.php
+++ b/Integration/Connector/CategoryConnector.php
@@ -17,7 +17,7 @@ class CategoryConnector extends AbstractConnector
     /**
      * {@inheritdoc}
      */
-    public function getLabel()
+    public function getLabel(): string
     {
         return 'oro.akeneo.connector.category.label';
     }
diff --git a/Integration/Connector/ProductConnector.php b/Integration/Connector/ProductConnector.php
index 69818fdf..9f65eb76 100644
--- a/Integration/Connector/ProductConnector.php
+++ b/Integration/Connector/ProductConnector.php
@@ -26,7 +26,7 @@ class ProductConnector extends AbstractConnector implements ConnectorInterface,
     /**
      * {@inheritdoc}
      */
-    public function getLabel()
+    public function getLabel(): string
     {
         return 'oro.akeneo.connector.product.label';
     }
diff --git a/Job/Context/SimpleContextAggregator.php b/Job/Context/SimpleContextAggregator.php
index 981cf7aa..c98e8a7e 100644
--- a/Job/Context/SimpleContextAggregator.php
+++ b/Job/Context/SimpleContextAggregator.php
@@ -2,7 +2,7 @@
 
 namespace Oro\Bundle\AkeneoBundle\Job\Context;
 
-use Akeneo\Bundle\BatchBundle\Entity\JobExecution;
+use Oro\Bundle\BatchBundle\Entity\JobExecution;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 use Oro\Bundle\ImportExportBundle\Job\Context\SimpleContextAggregator as BaseAggregator;
 use Oro\Bundle\ImportExportBundle\Job\ContextHelper;
diff --git a/Migrations/Schema/v1_8/OroAkeneoMigration.php b/Migrations/Schema/v1_8/OroAkeneoMigration.php
index 8019a6c5..8bf8f541 100644
--- a/Migrations/Schema/v1_8/OroAkeneoMigration.php
+++ b/Migrations/Schema/v1_8/OroAkeneoMigration.php
@@ -3,6 +3,7 @@
 namespace Oro\Bundle\AkeneoBundle\Migrations\Schema\v1_8;
 
 use Doctrine\DBAL\Schema\Schema;
+use Oro\Bundle\EntityBundle\Provider\EntityFieldProvider;
 use Oro\Bundle\EntityConfigBundle\Migration\UpdateEntityConfigFieldValueQuery;
 use Oro\Bundle\MigrationBundle\Migration\Migration;
 use Oro\Bundle\MigrationBundle\Migration\QueryBag;
@@ -18,7 +19,7 @@ public function up(Schema $schema, QueryBag $queries)
     {
         $fields = $this->container
             ->get('oro_entity.helper.field_helper')
-            ->getFields(Product::class, true);
+            ->getEntityFields(Product::class, EntityFieldProvider::OPTION_WITH_RELATIONS);
 
         $importExportProvider = $this->container
             ->get('oro_entity_config.config_manager')
diff --git a/Resources/config/batch_jobs.yml b/Resources/config/batch_jobs.yml
index 75fc8a5b..3a42d5b7 100644
--- a/Resources/config/batch_jobs.yml
+++ b/Resources/config/batch_jobs.yml
@@ -78,13 +78,13 @@ connector:
             reader: oro_akeneo.importexport.reader.product
             processor: oro_akeneo.importexport.processor.product
             writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
-        import_images:
-          title: import
-          class: Oro\Bundle\AkeneoBundle\ImportExport\Step\ItemStep
-          services:
-            reader: oro_akeneo.importexport.reader.product_image
-            processor: oro_akeneo.importexport.processor.product_image
-            writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
+#        import_images:
+#          title: import
+#          class: Oro\Bundle\AkeneoBundle\ImportExport\Step\ItemStep
+#          services:
+#            reader: oro_akeneo.importexport.reader.product_image
+#            processor: oro_akeneo.importexport.processor.product_image
+#            writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
         import_prices:
           title: import
           class: Oro\Bundle\BatchBundle\Step\ItemStep
diff --git a/Resources/config/controllers.yml b/Resources/config/controllers.yml
new file mode 100644
index 00000000..7781e9b5
--- /dev/null
+++ b/Resources/config/controllers.yml
@@ -0,0 +1,13 @@
+services:
+    _defaults:
+        public: true
+
+    Oro\Bundle\AkeneoBundle\Controller\ValidateConnectionController:
+        arguments:
+            - '@oro_currency.config.currency'
+            - '@translator'
+            - '@oro_akeneo.integration.transport'
+        calls:
+            - [setContainer, ['@Psr\Container\ContainerInterface']]
+        tags:
+            - { name: container.service_subscriber }
diff --git a/Resources/config/oro/placeholders.yml b/Resources/config/oro/placeholders.yml
index d8487f91..8f4622d6 100644
--- a/Resources/config/oro/placeholders.yml
+++ b/Resources/config/oro/placeholders.yml
@@ -1,7 +1,7 @@
 placeholders:
   items:
     akeneo_schema_update:
-      template: OroAkeneoBundle:Button:schema_update.html.twig
+      template: '@@OroAkeneo/Button/schema_update.html.twig'
       applicable: '@oro_akeneo.placeholder.schema_update_filter->isApplicable($entity$, Oro\Bundle\ProductBundle\Entity\Product)'
       data:
         entity_id: '@oro_entity_config.config_manager->getConfigModelId(Oro\Bundle\ProductBundle\Entity\Product)'
diff --git a/Resources/config/oro/twig.yml b/Resources/config/oro/twig.yml
index a15e3e79..3886f00f 100644
--- a/Resources/config/oro/twig.yml
+++ b/Resources/config/oro/twig.yml
@@ -1,2 +1,2 @@
 bundles:
-  - OroAkeneoBundle:Form:fields.html.twig
+  - '@OroAkeneo/Form/fields.html.twig'
diff --git a/Resources/config/services.yml b/Resources/config/services.yml
index 7b6ff9bc..26a5acd5 100644
--- a/Resources/config/services.yml
+++ b/Resources/config/services.yml
@@ -139,13 +139,17 @@ services:
     arguments:
       - '@oro_akeneo.event_listener.doctrine_tag.decorator.inner'
 
-  oro_akeneo.event_listener.import_export_tags_subscriber.decorator:
-    class: Oro\Bundle\AkeneoBundle\EventListener\ImportExportTagsSubscriberDecorator
-    decorates: oro_tag.event_listener.import_export_tags_subscriber
-    decoration_priority: -255
-    public: true
-    arguments:
-      - '@oro_akeneo.event_listener.import_export_tags_subscriber.decorator.inner'
+#  oro_akeneo.event_listener.import_export_tags_subscriber.decorator:
+#    class: Oro\Bundle\AkeneoBundle\EventListener\ImportExportTagsSubscriberDecorator
+#    decorates: oro_tag.event_listener.import_export_tags_subscriber
+#    decoration_priority: -255
+#    public: true
+#    arguments:
+#      - '@oro_akeneo.event_listener.import_export_tags_subscriber.decorator.inner'
+#
+#  oro_akeneo.tag_import.manager:
+#    alias: oro_tag.tag_import.manager
+#    public: true
 
   oro_akeneo.event_listener.load_class_metadata:
     class: Oro\Bundle\AkeneoBundle\EventListener\LoadClassMetadataListener
diff --git a/Resources/views/Button/schema_update.html.twig b/Resources/views/Button/schema_update.html.twig
index fef181ad..77705d0c 100644
--- a/Resources/views/Button/schema_update.html.twig
+++ b/Resources/views/Button/schema_update.html.twig
@@ -1,4 +1,4 @@
-{% import 'OroEntityConfigBundle::macros.html.twig' as entityConfig %}
+{% import '@OroEntityConfig/macros.html.twig' as entityConfig %}
 
 {% if is_granted('oro_entityconfig_manage') %}
     {% set button_config = [{
diff --git a/Resources/views/Datagrid/attributeFamilies.html.twig b/Resources/views/Datagrid/attributeFamilies.html.twig
index 48bc466d..b329b0c0 100644
--- a/Resources/views/Datagrid/attributeFamilies.html.twig
+++ b/Resources/views/Datagrid/attributeFamilies.html.twig
@@ -1,4 +1,4 @@
-{% import 'OroUIBundle::macros.html.twig' as UI %}
+{% import '@OroUI/macros.html.twig' as UI %}
 
 {% if is_granted('oro_attribute_family_view') %}
 <ul class="extra-list">
diff --git a/Resources/views/Form/fields.html.twig b/Resources/views/Form/fields.html.twig
index e4334896..5a09ef44 100644
--- a/Resources/views/Form/fields.html.twig
+++ b/Resources/views/Form/fields.html.twig
@@ -17,7 +17,7 @@
 {% endblock %}
 
 {% block oro_akeneo_settings_widget %}
-    {% import 'OroUIBundle::macros.html.twig' as UI %}
+    {% import '@OroUI/macros.html.twig' as UI %}
 
     <div class="akeneo-settings">
         <div data-page-component-module="oroakeneo/js/app/components/akeneo-transport-settings-component"
diff --git a/composer.json b/composer.json
index e24dd127..d66474c5 100644
--- a/composer.json
+++ b/composer.json
@@ -18,7 +18,7 @@
         ]
     },
     "require": {
-        "oro/commerce": "4.2.*",
+        "oro/commerce": "5.0.*",
         "akeneo/api-php-client-ee": "5.* || 6.*"
     }
 }

From 746fbf2e83e560d94f7243d38539192d004e4b8f Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <1804871+dxops@users.noreply.github.com>
Date: Sun, 6 Mar 2022 23:58:35 +0200
Subject: [PATCH 06/14] AKM-32: Move data from messages/RMQ to jobs/Database
 (#62)

---
 .github/workflows/php.yml                    |  20 ++
 .php_cs.php                                  |  23 +-
 .travis.yml                                  |  12 -
 Async/ImportProductProcessor.php             |  24 +-
 Command/CleanupCommand.php                   | 104 +++++++++
 DependencyInjection/OroAkeneoExtension.php   |   1 +
 ImportExport/Reader/ProductImageReader.php   |   7 +-
 ImportExport/Reader/ProductPriceReader.php   |   8 +-
 ImportExport/Reader/ProductReader.php        |   8 +-
 ImportExport/Reader/ProductVariantReader.php |   8 +-
 ImportExport/Writer/AsyncWriter.php          | 221 +++++++++----------
 Integration/Connector/ProductConnector.php   |  15 +-
 Resources/config/commands.yml                |   9 +
 Resources/config/importexport.yml            |  10 +-
 Resources/config/services.yml                |   2 +
 Tools/CacheProviderTrait.php                 |  16 ++
 16 files changed, 323 insertions(+), 165 deletions(-)
 create mode 100644 .github/workflows/php.yml
 delete mode 100644 .travis.yml
 create mode 100644 Command/CleanupCommand.php
 create mode 100644 Resources/config/commands.yml
 create mode 100644 Tools/CacheProviderTrait.php

diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml
new file mode 100644
index 00000000..a3b87ee7
--- /dev/null
+++ b/.github/workflows/php.yml
@@ -0,0 +1,20 @@
+name: PHP Composer
+
+on:
+  push:
+    branches: [ master 4.2 4.1 3.1 1.6 ]
+  pull_request:
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v2
+
+    - name: Install dependencies
+      run: php -r "copy('https://cs.symfony.com/download/php-cs-fixer-v3.phar', 'php-cs-fixer.phar');"
+      
+    - name: Run php-cs-fixer 
+      run: php php-cs-fixer.phar fix --dry-run --config=.php_cs.php --cache-file=.php_cs.cache --verbose --show-progress=dots --diff --allow-risky=yes
diff --git a/.php_cs.php b/.php_cs.php
index 9c9bd589..010fef1c 100644
--- a/.php_cs.php
+++ b/.php_cs.php
@@ -1,17 +1,18 @@
 <?php
 
-$finder = PhpCsFixer\Finder::create()
+$finder = \PhpCsFixer\Finder::create()
     ->in(__DIR__);
 
-return PhpCsFixer\Config::create()
+return (new \PhpCsFixer\Config())
     ->setFinder($finder)
     ->setRules(
         [
             // generic PSRs
             '@PSR1' => true,
             '@PSR2' => true,
-            'psr0' => true,
-            'psr4' => true,
+            '@PSR12' => true,
+            '@PSR12:risky' => true,
+            'psr_autoloading' => true,
 
             // imports
             'ordered_imports' => true,
@@ -39,11 +40,14 @@
             'escape_implicit_backslashes' => false,
 
             // PHP
-            '@PHP71Migration' => true,
-            '@PHP71Migration:risky' => true,
+            '@PHP74Migration' => true,
+            '@PHP74Migration:risky' => true,
+            '@PHP80Migration' => true,
+            '@PHP80Migration:risky' => true,
 
+            'use_arrow_functions' => false,
+            'get_class_to_class_keyword' => false,
             'void_return' => false,
-            'visibility_required' => false,
             'list_syntax' => ['syntax' => 'long'],
             'declare_strict_types' => false,
 
@@ -64,11 +68,11 @@
             '@Symfony:risky' => true,
             'phpdoc_types_order' => false,
             'phpdoc_separation' => false,
-            'phpdoc_inline_tag' => false,
+            'visibility_required' => ['elements' => ['property', 'method']],
+            'types_spaces' => false,
             'native_function_invocation' => false,
             'concat_space' => ['spacing' => 'one'],
             'single_space_after_construct' => false,
-            'trailing_comma_in_multiline_array' => false,
             'self_accessor' => false,
             'yoda_style' => false,
             'phpdoc_summary' => false,
@@ -87,6 +91,7 @@
             'ternary_operator_spaces' => false,
             'phpdoc_no_useless_inheritdoc' => false,
             'class_definition' => false,
+            'string_length_to_empty' => false,
         ]
     )
     ->setRiskyAllowed(true);
diff --git a/.travis.yml b/.travis.yml
deleted file mode 100644
index 40ecb0f3..00000000
--- a/.travis.yml
+++ /dev/null
@@ -1,12 +0,0 @@
-language: php
-
-php:
-    - '8.0'
-
-install:
-    - php -r "copy('https://cs.symfony.com/download/php-cs-fixer-v2.phar', 'php-cs-fixer.phar');"
-    - php -r "copy('https://squizlabs.github.io/PHP_CodeSniffer/phpcs.phar', 'phpcs.phar');"
-
-script:
-    - php php-cs-fixer.phar fix --dry-run --config=.php_cs.php --cache-file=.php_cs.cache --verbose --show-progress=dots --diff --allow-risky=yes
-    - php phpcs.phar -p --extensions=php --standard=PSR1,PSR12 --cache=.phpcs.cache --report=code --report=diff -n .
diff --git a/Async/ImportProductProcessor.php b/Async/ImportProductProcessor.php
index b023086c..b56dd02d 100644
--- a/Async/ImportProductProcessor.php
+++ b/Async/ImportProductProcessor.php
@@ -3,13 +3,15 @@
 namespace Oro\Bundle\AkeneoBundle\Async;
 
 use Doctrine\ORM\EntityManagerInterface;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
 use Oro\Bundle\IntegrationBundle\Authentication\Token\IntegrationTokenAwareTrait;
 use Oro\Bundle\IntegrationBundle\Entity\Channel as Integration;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
 use Oro\Bundle\IntegrationBundle\Provider\SyncProcessorRegistry;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
 use Oro\Component\MessageQueue\Client\TopicSubscriberInterface;
 use Oro\Component\MessageQueue\Consumption\MessageProcessorInterface;
-use Oro\Component\MessageQueue\Job\Job;
 use Oro\Component\MessageQueue\Job\JobRunner;
 use Oro\Component\MessageQueue\Transport\MessageInterface;
 use Oro\Component\MessageQueue\Transport\SessionInterface;
@@ -19,6 +21,7 @@
 
 class ImportProductProcessor implements MessageProcessorInterface, TopicSubscriberInterface
 {
+    use CacheProviderTrait;
     use IntegrationTokenAwareTrait;
 
     /** @var DoctrineHelper */
@@ -97,12 +100,31 @@ public function process(MessageInterface $message, SessionInterface $session)
             function (JobRunner $jobRunner, Job $child) use ($integration, $body) {
                 $this->doctrineHelper->refreshIncludingUnitializedRelations($integration);
                 $processor = $this->syncProcessorRegistry->getProcessorForIntegration($integration);
+
+                $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
+                /** @var FieldsChanges $fieldsChanges */
+                $fieldsChanges = $em
+                    ->getRepository(FieldsChanges::class)
+                    ->findOneBy(['entityId' => $child->getId(), 'entityClass' => Job::class]);
+
+                if (!$fieldsChanges) {
+                    $this->logger->error(
+                        sprintf('Source data from Akeneo not found for job: %s', $child->getId())
+                    );
+
+                    return false;
+                }
+
+                $this->cacheProvider->save('akeneo', $fieldsChanges->getChangedFields());
+
                 $status = $processor->process(
                     $integration,
                     $body['connector'] ?? null,
                     $body['connector_parameters'] ?? []
                 );
 
+                $em->clear(FieldsChanges::class);
+
                 return $status;
             }
         );
diff --git a/Command/CleanupCommand.php b/Command/CleanupCommand.php
new file mode 100644
index 00000000..d4e2e794
--- /dev/null
+++ b/Command/CleanupCommand.php
@@ -0,0 +1,104 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Command;
+
+use Oro\Bundle\BatchBundle\ORM\Query\BufferedIdentityQueryResultIterator;
+use Oro\Bundle\CronBundle\Command\CronCommandInterface;
+use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * Clears old records from oro_integration_fields_changes table before repack.
+ */
+class CleanupCommand extends Command implements CronCommandInterface
+{
+    /** @var string */
+    protected static $defaultName = 'oro:akeneo:cleanup';
+
+    /** @var DoctrineHelper */
+    private $doctrineHelper;
+
+    public function __construct(DoctrineHelper $doctrineHelper)
+    {
+        $this->doctrineHelper = $doctrineHelper;
+        parent::__construct();
+    }
+
+    public function isActive()
+    {
+        return true;
+    }
+
+    public function getDefaultDefinition()
+    {
+        return '0 2 * * 6';
+    }
+
+    public function configure()
+    {
+        $this
+            ->setDescription('Clears old records from oro_integration_fields_changes table.')
+            ->setHelp(
+                <<<'HELP'
+                    The <info>%command.name%</info> command clears fields changes for complete job records
+                    from <comment>oro_integration_fields_changes</comment> table.
+
+                      <info>php %command.full_name%</info>
+                    HELP
+            );
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $output->writeln(sprintf(
+            '<comment>Number of fields changes that has been deleted:</comment> %d',
+            $this->deleteRecords()
+        ));
+
+        $output->writeln('<info>Fields changes cleanup complete</info>');
+    }
+
+    private function deleteRecords(): int
+    {
+        $qb = $this->doctrineHelper
+            ->getEntityManagerForClass(FieldsChanges::class)
+            ->getRepository(FieldsChanges::class)
+            ->createQueryBuilder('fc');
+
+        $qb
+            ->delete(FieldsChanges::class, 'fc')
+            ->where($qb->expr()->eq('fc.entityClass', ':class'))
+            ->setParameter('class', Job::class)
+            ->andWhere($qb->expr()->in('fc.entityId', ':ids'));
+
+        $jqb = $this->doctrineHelper
+            ->getEntityManagerForClass(Job::class)
+            ->getRepository(Job::class)
+            ->createQueryBuilder('j');
+
+        $jqb
+            ->select('j.id')
+            ->where($jqb->expr()->in('j.status', ':statuses'))
+            ->setParameter('statuses', [Job::STATUS_SUCCESS, Job::STATUS_CANCELLED, Job::STATUS_FAILED, Job::STATUS_STALE])
+            ->orderBy($jqb->expr()->desc('j.id'));
+
+        $iterator = new BufferedIdentityQueryResultIterator($jqb->getQuery());
+
+        $result = 0;
+        $iterator->setPageLoadedCallback(function (array $rows) use ($qb, &$result): array {
+            $ids = array_column($rows, 'id');
+
+            $result = $result + $qb->setParameter('ids', $ids)->getQuery()->execute();
+
+            return $ids;
+        });
+
+        iterator_to_array($iterator);
+
+        return $result;
+    }
+}
diff --git a/DependencyInjection/OroAkeneoExtension.php b/DependencyInjection/OroAkeneoExtension.php
index fcf59856..c2c0972a 100644
--- a/DependencyInjection/OroAkeneoExtension.php
+++ b/DependencyInjection/OroAkeneoExtension.php
@@ -22,6 +22,7 @@ class OroAkeneoExtension extends Extension
     public function load(array $configs, ContainerBuilder $container)
     {
         $loader = new Loader\YamlFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config'));
+        $loader->load('commands.yml');
         $loader->load('integration.yml');
         $loader->load('controllers.yml');
         $loader->load('importexport.yml');
diff --git a/ImportExport/Reader/ProductImageReader.php b/ImportExport/Reader/ProductImageReader.php
index 5435aa98..265f629e 100644
--- a/ImportExport/Reader/ProductImageReader.php
+++ b/ImportExport/Reader/ProductImageReader.php
@@ -4,12 +4,14 @@
 
 use Oro\Bundle\AkeneoBundle\ImportExport\AkeneoIntegrationTrait;
 use Oro\Bundle\AkeneoBundle\Integration\AkeneoFileManager;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 
 class ProductImageReader extends IteratorBasedReader
 {
     use AkeneoIntegrationTrait;
+    use CacheProviderTrait;
 
     /** @var array */
     private $attributesImageFilter = [];
@@ -45,10 +47,7 @@ protected function initializeFromContext(ContextInterface $context)
 
         $this->initAttributesImageList();
 
-        $items = $this->stepExecution
-                ->getJobExecution()
-                ->getExecutionContext()
-                ->get('items') ?? [];
+        $items = $this->cacheProvider->fetch('akeneo')['items'] ?? [];
 
         if (!empty($items)) {
             $this->processImagesDownload($items, $context);
diff --git a/ImportExport/Reader/ProductPriceReader.php b/ImportExport/Reader/ProductPriceReader.php
index 7d71d917..14038ac4 100644
--- a/ImportExport/Reader/ProductPriceReader.php
+++ b/ImportExport/Reader/ProductPriceReader.php
@@ -2,18 +2,18 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Reader;
 
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 
 class ProductPriceReader extends IteratorBasedReader
 {
+    use CacheProviderTrait;
+
     protected function initializeFromContext(ContextInterface $context)
     {
         parent::initializeFromContext($context);
 
-        $items = $this->stepExecution
-                ->getJobExecution()
-                ->getExecutionContext()
-                ->get('items') ?? [];
+        $items = $this->cacheProvider->fetch('akeneo')['items'] ?? [];
 
         $prices = [];
 
diff --git a/ImportExport/Reader/ProductReader.php b/ImportExport/Reader/ProductReader.php
index 4605d2a9..3c4ce8ac 100644
--- a/ImportExport/Reader/ProductReader.php
+++ b/ImportExport/Reader/ProductReader.php
@@ -3,10 +3,13 @@
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Reader;
 
 use Oro\Bundle\AkeneoBundle\Integration\AkeneoFileManager;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 
 class ProductReader extends IteratorBasedReader
 {
+    use CacheProviderTrait;
+
     /** @var AkeneoFileManager */
     private $akeneoFileManager;
 
@@ -19,10 +22,7 @@ protected function initializeFromContext(ContextInterface $context)
     {
         parent::initializeFromContext($context);
 
-        $items = $this->stepExecution
-                ->getJobExecution()
-                ->getExecutionContext()
-                ->get('items') ?? [];
+        $items = $this->cacheProvider->fetch('akeneo')['items'] ?? [];
 
         if (!empty($items)) {
             $this->processFileTypeDownload($items, $context);
diff --git a/ImportExport/Reader/ProductVariantReader.php b/ImportExport/Reader/ProductVariantReader.php
index 4ea8f36f..b28d4379 100644
--- a/ImportExport/Reader/ProductVariantReader.php
+++ b/ImportExport/Reader/ProductVariantReader.php
@@ -2,18 +2,18 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Reader;
 
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\ImportExportBundle\Context\ContextInterface;
 
 class ProductVariantReader extends IteratorBasedReader
 {
+    use CacheProviderTrait;
+
     protected function initializeFromContext(ContextInterface $context)
     {
         parent::initializeFromContext($context);
 
-        $variants = $this->stepExecution
-                ->getJobExecution()
-                ->getExecutionContext()
-                ->get('variants') ?? [];
+        $variants = $this->cacheProvider->fetch('akeneo')['variants'] ?? [];
 
         $this->stepExecution->setReadCount(count($variants));
 
diff --git a/ImportExport/Writer/AsyncWriter.php b/ImportExport/Writer/AsyncWriter.php
index e223401d..bf88761a 100644
--- a/ImportExport/Writer/AsyncWriter.php
+++ b/ImportExport/Writer/AsyncWriter.php
@@ -2,29 +2,30 @@
 
 namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer;
 
-use Doctrine\Common\Cache\CacheProvider;
+use Doctrine\DBAL\Platforms\MySqlPlatform;
+use Doctrine\DBAL\Types\Types;
 use Oro\Bundle\AkeneoBundle\Async\Topics;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\BatchBundle\Entity\StepExecution;
 use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
 use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
 use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
+use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
 use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
 use Oro\Component\MessageQueue\Client\Message;
 use Oro\Component\MessageQueue\Client\MessagePriority;
 use Oro\Component\MessageQueue\Client\MessageProducerInterface;
-use Oro\Component\MessageQueue\Job\Job;
-use Oro\Component\MessageQueue\Job\JobProcessor;
-use Oro\Component\MessageQueue\Job\JobRunner;
 
 class AsyncWriter implements
     ItemWriterInterface,
     ClosableInterface,
     StepExecutionAwareInterface
 {
-    private const VARIANTS_BATCH_SIZE = 25;
+    use CacheProviderTrait;
 
-    /** @var JobRunner */
-    private $jobRunner;
+    private const VARIANTS_BATCH_SIZE = 25;
 
     /** @var MessageProducerInterface * */
     private $messageProducer;
@@ -32,31 +33,22 @@ class AsyncWriter implements
     /** @var StepExecution */
     private $stepExecution;
 
-    /** @var int */
-    private $key = 0;
-
     /** @var int */
     private $size = 0;
 
-    /** @var CacheProvider */
-    private $cacheProvider;
-
-    /** @var JobProcessor */
-    private $jobProcessor;
+    /** @var DoctrineHelper */
+    private $doctrineHelper;
 
     public function __construct(
-        JobRunner $jobRunner,
         MessageProducerInterface $messageProducer,
-        JobProcessor $jobProcessor
+        DoctrineHelper $doctrineHelper
     ) {
-        $this->jobRunner = $jobRunner;
         $this->messageProducer = $messageProducer;
-        $this->jobProcessor = $jobProcessor;
+        $this->doctrineHelper = $doctrineHelper;
     }
 
     public function initialize()
     {
-        $this->key = 1;
         $this->size = 0;
     }
 
@@ -72,56 +64,15 @@ public function write(array $items)
             $newSize
         );
         $this->size = $newSize;
-
         $this->stepExecution->setWriteCount($this->size);
 
-        $setRootJob = \Closure::bind(
-            function ($property, $value) {
-                $this->{$property} = $value;
-            },
-            $this->jobRunner,
-            $this->jobRunner
-        );
-
-        try {
-            $setRootJob('rootJob', $this->getRootJob());
-
-            $this->jobRunner->createDelayed(
-                $jobName,
-                function (JobRunner $jobRunner, Job $child) use ($items, $channelId) {
-                    $this->messageProducer->send(
-                        Topics::IMPORT_PRODUCTS,
-                        new Message(
-                            [
-                                'integrationId' => $channelId,
-                                'jobId' => $child->getId(),
-                                'connector' => 'product',
-                                'connector_parameters' => [
-                                    'items' => $items,
-                                    'incremented_read' => true,
-                                ],
-                            ],
-                            MessagePriority::HIGH
-                        )
-                    );
-
-                    if ($this->messageProducer instanceof BufferedMessageProducer && $this->messageProducer->isBufferingEnabled()) {
-                        $this->messageProducer->flushBuffer();
-                    }
-
-                    return true;
-                }
-            );
-        } finally {
-            $setRootJob('rootJob', null);
-
-            $this->key++;
-        }
+        $jobId = $this->insertJob($jobName);
+        $this->createFieldsChanges($jobId, $items, 'items');
+        $this->sendMessage($channelId, $jobId, true);
     }
 
     public function flush()
     {
-        $this->key = 1;
         $this->size = 0;
 
         $variants = $this->cacheProvider->fetch('product_variants') ?? [];
@@ -131,70 +82,72 @@ public function flush()
 
         $channelId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('channel');
 
-        $setRootJob = \Closure::bind(
-            function ($property, $value) {
-                $this->{$property} = $value;
-            },
-            $this->jobRunner,
-            $this->jobRunner
+        $chunks = array_chunk($variants, self::VARIANTS_BATCH_SIZE, true);
+
+        foreach ($chunks as $key => $chunk) {
+            $jobName = sprintf(
+                'oro_integration:sync_integration:%s:variants:%s-%s',
+                $channelId,
+                self::VARIANTS_BATCH_SIZE * $key + 1,
+                self::VARIANTS_BATCH_SIZE * $key + count($chunk)
+            );
+
+            $jobId = $this->insertJob($jobName);
+            $this->createFieldsChanges($jobId, $chunk, 'variants');
+            $this->sendMessage($channelId, $jobId);
+        }
+    }
+
+    private function createFieldsChanges(int $jobId, array &$data, string $key): void
+    {
+        $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
+        $fieldsChanges = $em
+            ->getRepository(FieldsChanges::class)
+            ->findOneBy(['entityId' => $jobId, 'entityClass' => Job::class]);
+        if (!$fieldsChanges) {
+            $fieldsChanges = new FieldsChanges([]);
+            $fieldsChanges->setEntityClass(Job::class);
+            $fieldsChanges->setEntityId($jobId);
+        }
+        $fieldsChanges->setChangedFields([$key => $data]);
+        $em->persist($fieldsChanges);
+        $em->flush($fieldsChanges);
+        $em->clear(FieldsChanges::class);
+    }
+
+    private function sendMessage(int $channelId, int $jobId, bool $incrementedRead = false): void
+    {
+        $this->messageProducer->send(
+            Topics::IMPORT_PRODUCTS,
+            new Message(
+                [
+                    'integrationId' => $channelId,
+                    'jobId' => $jobId,
+                    'connector' => 'product',
+                    'connector_parameters' => ['incremented_read' => $incrementedRead],
+                ],
+                MessagePriority::HIGH
+            )
         );
 
-        try {
-            $setRootJob('rootJob', $this->getRootJob());
-            $chunks = array_chunk($variants, self::VARIANTS_BATCH_SIZE, true);
-
-            foreach ($chunks as $key => $chunk) {
-                $jobName = sprintf(
-                    'oro_integration:sync_integration:%s:variants:%s-%s',
-                    $channelId,
-                    self::VARIANTS_BATCH_SIZE * $key + 1,
-                    self::VARIANTS_BATCH_SIZE * $key + count($chunk)
-                );
-                $this->jobRunner->createDelayed(
-                    $jobName,
-                    function (JobRunner $jobRunner, Job $child) use ($channelId, $chunk) {
-                        $this->messageProducer->send(
-                            Topics::IMPORT_PRODUCTS,
-                            new Message(
-                                [
-                                    'integrationId' => $channelId,
-                                    'jobId' => $child->getId(),
-                                    'connector' => 'product',
-                                    'connector_parameters' => [
-                                        'variants' => $chunk,
-                                    ],
-                                ],
-                                MessagePriority::HIGH
-                            )
-                        );
-
-                        return true;
-                    }
-                );
-            }
-        } finally {
-            $setRootJob('rootJob', null);
+        if ($this->messageProducer instanceof BufferedMessageProducer
+            && $this->messageProducer->isBufferingEnabled()) {
+            $this->messageProducer->flushBuffer();
         }
     }
 
-    private function getRootJob(): Job
+    private function getRootJob(): ?int
     {
         $rootJobId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('rootJobId') ?? null;
         if (!$rootJobId) {
             throw new \InvalidArgumentException('Root job id is empty');
         }
 
-        $rootJob = $this->jobProcessor->findJobById($rootJobId);
-        if (!$rootJob) {
-            throw new \InvalidArgumentException('Root job is empty');
-        }
-
-        return $rootJob;
+        return $rootJobId;
     }
 
     public function close()
     {
-        $this->key = 1;
         $this->size = 0;
     }
 
@@ -203,8 +156,46 @@ public function setStepExecution(StepExecution $stepExecution)
         $this->stepExecution = $stepExecution;
     }
 
-    public function setCacheProvider(CacheProvider $cacheProvider): void
+    private function insertJob(string $jobName): ?int
     {
-        $this->cacheProvider = $cacheProvider;
+        $em = $this->doctrineHelper->getEntityManager(Job::class);
+        $tableName = $em->getClassMetadata(Job::class)->getTableName();
+        $connection = $em->getConnection();
+
+        $qb = $connection->createQueryBuilder();
+        $qb
+            ->insert($tableName)
+            ->values([
+                'name' => ':name',
+                'status' => ':status',
+                'interrupted' => ':interrupted',
+                'created_at' => ':createdAt',
+                'root_job_id' => ':rootJob',
+            ])
+            ->setParameters([
+                'name' => $jobName,
+                'status' => Job::STATUS_NEW,
+                'interrupted' => false,
+                'unique' => false,
+                'createdAt' => new \DateTime(),
+                'rootJob' => $this->getRootJob(),
+            ], [
+                'name' => Types::STRING,
+                'status' => Types::STRING,
+                'interrupted' => Types::BOOLEAN,
+                'unique' => Types::BOOLEAN,
+                'createdAt' => Types::DATETIME_MUTABLE,
+                'rootJob' => Types::INTEGER,
+            ]);
+
+        if ($connection->getDatabasePlatform() instanceof MySqlPlatform) {
+            $qb->setValue('`unique`', ':unique');
+        } else {
+            $qb->setValue('"unique"', ':unique');
+        }
+
+        $qb->execute();
+
+        return $connection->lastInsertId();
     }
 }
diff --git a/Integration/Connector/ProductConnector.php b/Integration/Connector/ProductConnector.php
index 9f65eb76..95bd70f4 100644
--- a/Integration/Connector/ProductConnector.php
+++ b/Integration/Connector/ProductConnector.php
@@ -3,6 +3,7 @@
 namespace Oro\Bundle\AkeneoBundle\Integration\Connector;
 
 use Oro\Bundle\AkeneoBundle\Placeholder\SchemaUpdateFilter;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
 use Oro\Bundle\IntegrationBundle\Entity\Channel;
 use Oro\Bundle\IntegrationBundle\Provider\AbstractConnector;
 use Oro\Bundle\IntegrationBundle\Provider\AllowedConnectorInterface;
@@ -14,6 +15,8 @@
  */
 class ProductConnector extends AbstractConnector implements ConnectorInterface, AllowedConnectorInterface
 {
+    use CacheProviderTrait;
+
     const IMPORT_JOB_NAME = 'akeneo_product_import';
     const PAGE_SIZE = 100;
     const TYPE = 'product';
@@ -73,20 +76,12 @@ public function setSchemaUpdateFilter(SchemaUpdateFilter $schemaUpdateFilter): v
      */
     protected function getConnectorSource()
     {
-        $items = $this->stepExecution
-            ->getJobExecution()
-            ->getExecutionContext()
-            ->get('items');
-
+        $items = $this->cacheProvider->fetch('akeneo')['items'] ?? [];
         if ($items) {
             return new \ArrayIterator();
         }
 
-        $variants = $this->stepExecution
-            ->getJobExecution()
-            ->getExecutionContext()
-            ->get('variants');
-
+        $variants = $this->cacheProvider->fetch('akeneo')['variants'] ?? [];
         if ($variants) {
             return new \ArrayIterator();
         }
diff --git a/Resources/config/commands.yml b/Resources/config/commands.yml
new file mode 100644
index 00000000..60d55f6d
--- /dev/null
+++ b/Resources/config/commands.yml
@@ -0,0 +1,9 @@
+services:
+    _defaults:
+        public: false
+
+    Oro\Bundle\AkeneoBundle\Command\CleanupCommand:
+        arguments:
+            - '@oro_entity.doctrine_helper'
+        tags:
+            - { name: console.command }
diff --git a/Resources/config/importexport.yml b/Resources/config/importexport.yml
index 83ae9f09..2243189c 100644
--- a/Resources/config/importexport.yml
+++ b/Resources/config/importexport.yml
@@ -210,6 +210,7 @@ services:
       - '@oro_integration.provider.connector_context_mediator'
     calls:
       - [ setSchemaUpdateFilter, [ '@oro_akeneo.placeholder.schema_update_filter' ] ]
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
     tags:
       - { name: oro_integration.connector, type: product, channel_type: oro_akeneo }
 
@@ -312,9 +313,8 @@ services:
   oro_akeneo.importexport.writer.async_product:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Writer\AsyncWriter'
     arguments:
-      - '@oro_message_queue.job.runner'
       - '@oro_message_queue.message_producer'
-      - '@oro_message_queue.job.processor'
+      - '@oro_entity.doctrine_helper'
     calls:
       - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
@@ -336,11 +336,14 @@ services:
       - '@oro_importexport.context_registry'
     calls:
       - [ setAkeneoFileManager, [ '@oro_akeneo.integration.akeneo_file_manager' ] ]
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.reader.price:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Reader\ProductPriceReader'
     arguments:
       - '@oro_importexport.context_registry'
+    calls:
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.reader.product_image:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Reader\ProductImageReader'
@@ -349,11 +352,14 @@ services:
     calls:
       - [ setDoctrineHelper, [ '@oro_entity.doctrine_helper' ] ]
       - [ setAkeneoFileManager, [ '@oro_akeneo.integration.akeneo_file_manager' ] ]
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.reader.product_variant:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Reader\ProductVariantReader'
     arguments:
       - '@oro_importexport.context_registry'
+    calls:
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.reader.category_parent:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Reader\CategoryParentReader'
diff --git a/Resources/config/services.yml b/Resources/config/services.yml
index 26a5acd5..b2704326 100644
--- a/Resources/config/services.yml
+++ b/Resources/config/services.yml
@@ -123,6 +123,8 @@ services:
       - '@security.token_storage'
       - '@logger'
       - '@oro_integration.processor_registry'
+    calls:
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
     tags:
       - { name: 'oro_message_queue.client.message_processor', topicName: 'oro.integration.akeneo.product' }
 
diff --git a/Tools/CacheProviderTrait.php b/Tools/CacheProviderTrait.php
new file mode 100644
index 00000000..6b1d51b8
--- /dev/null
+++ b/Tools/CacheProviderTrait.php
@@ -0,0 +1,16 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Tools;
+
+use Doctrine\Common\Cache\CacheProvider;
+
+trait CacheProviderTrait
+{
+    /** @var CacheProvider */
+    private $cacheProvider;
+
+    public function setCacheProvider(CacheProvider $cacheProvider): void
+    {
+        $this->cacheProvider = $cacheProvider;
+    }
+}

From a411656b3a4e55c06956305134b6fba96aec53c3 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Tue, 8 Mar 2022 20:40:12 +0200
Subject: [PATCH 07/14] AKM-33: OroCommerce 5 LTS compatibility

---
 Client/AkeneoClientFactory.php                |  6 ++--
 .../ImportexportFieldConfiguration.php        | 28 +++++++++++++++++++
 .../Schema/OroAkeneoBundleInstaller.php       |  1 -
 .../Schema/v1_12/OroAkeneoMigration.php       |  1 -
 README.md                                     | 20 ++++++-------
 Resources/config/importexport.yml             |  3 +-
 Resources/config/services.yml                 |  5 ++++
 composer.json                                 |  2 +-
 8 files changed, 47 insertions(+), 19 deletions(-)
 create mode 100644 EntityConfig/ImportexportFieldConfiguration.php

diff --git a/Client/AkeneoClientFactory.php b/Client/AkeneoClientFactory.php
index 6dd448e8..c2c9512c 100644
--- a/Client/AkeneoClientFactory.php
+++ b/Client/AkeneoClientFactory.php
@@ -2,8 +2,8 @@
 
 namespace Oro\Bundle\AkeneoBundle\Client;
 
+use Akeneo\Pim\ApiClient\AkeneoPimClientBuilder;
 use Akeneo\Pim\ApiClient\AkeneoPimClientInterface;
-use Akeneo\PimEnterprise\ApiClient\AkeneoPimEnterpriseClientBuilder;
 use Oro\Bundle\AkeneoBundle\Encoder\Crypter;
 use Oro\Bundle\AkeneoBundle\Entity\AkeneoSettings;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
@@ -117,9 +117,9 @@ private function createClientByToken(): AkeneoPimClientInterface
         return $this->client;
     }
 
-    private function getClientBuilder(): AkeneoPimEnterpriseClientBuilder
+    private function getClientBuilder(): AkeneoPimClientBuilder
     {
-        $clientBuilder = new AkeneoPimEnterpriseClientBuilder($this->akeneoUrl);
+        $clientBuilder = new AkeneoPimClientBuilder($this->akeneoUrl);
         $clientBuilder->setHttpClient($this->httpClient);
         $clientBuilder->setRequestFactory($this->requestFactory);
         $clientBuilder->setStreamFactory($this->streamFactory);
diff --git a/EntityConfig/ImportexportFieldConfiguration.php b/EntityConfig/ImportexportFieldConfiguration.php
new file mode 100644
index 00000000..1eb9ed2c
--- /dev/null
+++ b/EntityConfig/ImportexportFieldConfiguration.php
@@ -0,0 +1,28 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\EntityConfig;
+
+use Oro\Bundle\EntityConfigBundle\EntityConfig\FieldConfigInterface;
+use Symfony\Component\Config\Definition\Builder\NodeBuilder;
+
+/**
+ * Provides validations field config for importexport scope.
+ */
+class ImportexportFieldConfiguration implements FieldConfigInterface
+{
+    public function getSectionName(): string
+    {
+        return 'importexport';
+    }
+
+    public function configure(NodeBuilder $nodeBuilder): void
+    {
+        $nodeBuilder
+            ->scalarNode('source')
+                ->info('`string` source of field.')
+            ->end()
+            ->scalarNode('source_name')
+                ->info('`string` source name of field.')
+            ->end();
+    }
+}
diff --git a/Migrations/Schema/OroAkeneoBundleInstaller.php b/Migrations/Schema/OroAkeneoBundleInstaller.php
index b2ab9168..54aa4f57 100644
--- a/Migrations/Schema/OroAkeneoBundleInstaller.php
+++ b/Migrations/Schema/OroAkeneoBundleInstaller.php
@@ -21,7 +21,6 @@ class OroAkeneoBundleInstaller implements Installation, ExtendExtensionAwareInte
      */
     protected $options = [
         'extend' => [
-            'origin' => ExtendScope::OWNER_CUSTOM,
             'owner' => ExtendScope::OWNER_CUSTOM,
             'state' => ExtendScope::STATE_NEW,
             'is_serialized' => false,
diff --git a/Migrations/Schema/v1_12/OroAkeneoMigration.php b/Migrations/Schema/v1_12/OroAkeneoMigration.php
index f413b642..4f8003d5 100644
--- a/Migrations/Schema/v1_12/OroAkeneoMigration.php
+++ b/Migrations/Schema/v1_12/OroAkeneoMigration.php
@@ -43,7 +43,6 @@ public function up(Schema $schema, QueryBag $queries)
                 'notnull' => false,
                 'oro_options' => [
                     'extend' => [
-                        'origin' => ExtendScope::OWNER_CUSTOM,
                         'owner' => ExtendScope::OWNER_CUSTOM,
                         'state' => ExtendScope::STATE_NEW,
                         'is_serialized' => false,
diff --git a/README.md b/README.md
index f1aa0abc..e9a659bf 100644
--- a/README.md
+++ b/README.md
@@ -16,13 +16,13 @@ With this extension, you will be able to sync the following data from Akeneo to
 
 ## Compatibility
 
-| Connector | Status | OroCommerce |    Akeneo      | Build |
-|-----------|--------|-------------|----------------|-------|
-|    1.6    |  EOL   |     1.6     | 2.3, 3.2, 4.0* |       |
-|    3.1    |  EOL   |     3.1     | 2.3, 3.2, 4.0* |       |
-|    4.1    |  2021  |     4.1     | 2.3, 3.2, 4.0* | [![Build Status](https://travis-ci.org/oroinc/OroAkeneoBundle.svg?branch=4.1)](https://travis-ci.org/oroinc/OroAkeneoBundle) |
-|    4.2    |  2022  |     4.2     | 3.2, 4.0, 5.0  | [![Build Status](https://travis-ci.org/oroinc/OroAkeneoBundle.svg?branch=4.2)](https://travis-ci.org/oroinc/OroAkeneoBundle) |
-|    5.0    |  WIP   |     5.0     | 4.0, 5.0       |       |
+| Connector | Status | OroCommerce |    Akeneo      |
+|-----------|--------|-------------|----------------|
+|    1.6    |  EOL   |     1.6     | 2.3, 3.2, 4.0* |
+|    3.1    |  EOL   |     3.1     | 2.3, 3.2, 4.0* |
+|    4.1    |  EOL   |     4.1     | 2.3, 3.2, 4.0* |
+|    4.2    |  2022  |     4.2     | 3.2, 4.0, 5.0  |
+|    5.0    |  2023  |     5.0     | 5.0+           |
 
 ** Akeneo supported using older client versions, new features are not available.**
 
@@ -31,11 +31,7 @@ With this extension, you will be able to sync the following data from Akeneo to
 1. Add composer package
 
 ```
-# Akeneo 4.0
-composer require "oro/commerce-akeneo:4.2.*" "akeneo/api-php-client-ee:5.*"
-
-# Akeneo 5.0
-composer require "oro/commerce-akeneo:4.2.*" "akeneo/api-php-client-ee:6.*"
+composer require "oro/commerce-akeneo:5.0.*"
 ```
 
 2. Follow [Setup Guide](https://doc.oroinc.com/backend/setup/upgrade-to-new-version)
diff --git a/Resources/config/importexport.yml b/Resources/config/importexport.yml
index 2243189c..c6a34739 100644
--- a/Resources/config/importexport.yml
+++ b/Resources/config/importexport.yml
@@ -319,7 +319,8 @@ services:
       - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.cache:
-    class: Doctrine\Common\Cache\ArrayCache
+    parent: oro_cache.array_cache
+    public: false
     calls:
       - [ setNamespace, [ 'oro_akeneo' ] ]
 
diff --git a/Resources/config/services.yml b/Resources/config/services.yml
index b2704326..23b1b635 100644
--- a/Resources/config/services.yml
+++ b/Resources/config/services.yml
@@ -400,3 +400,8 @@ services:
     class: 'Oro\Bundle\AkeneoBundle\EventListener\DeletedAttributeRelationListener'
     decorates: oro_serialized_fields.event_listener.deleted_attribute_relation_serialized
     parent: oro_serialized_fields.event_listener.deleted_attribute_relation_serialized
+
+  oro_akeneo.entity_config.importexport_field_configuration:
+    class: Oro\Bundle\AkeneoBundle\EntityConfig\ImportexportFieldConfiguration
+    tags:
+      - oro_entity_config.validation.entity_config
diff --git a/composer.json b/composer.json
index d66474c5..8a435f87 100644
--- a/composer.json
+++ b/composer.json
@@ -19,6 +19,6 @@
     },
     "require": {
         "oro/commerce": "5.0.*",
-        "akeneo/api-php-client-ee": "5.* || 6.*"
+        "akeneo/api-php-client": "8.*"
     }
 }

From 5bc1d9cbf56a7d980f08ec5c5dfa9a23c6dadf61 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Wed, 9 Mar 2022 22:25:34 +0200
Subject: [PATCH 08/14] AKM-33: OroCommerce 5 LTS compatibility - fix 4.2 to
 5.0 settings migration

---
 Entity/AkeneoSettings.php        |  2 +-
 Form/Type/AkeneoLocaleType.php   | 12 ++++++++----
 Form/Type/AkeneoSettingsType.php | 22 +++++++++++++---------
 Integration/AkeneoTransport.php  |  4 ++--
 4 files changed, 24 insertions(+), 16 deletions(-)

diff --git a/Entity/AkeneoSettings.php b/Entity/AkeneoSettings.php
index 3fdf2828..3f7f6af9 100644
--- a/Entity/AkeneoSettings.php
+++ b/Entity/AkeneoSettings.php
@@ -523,7 +523,7 @@ public function setProductUnitPrecisionAttribute($productUnitPrecisionAttribute)
     }
 
     /**
-     * @return array/null
+     * @return array|null
      */
     public function getAkeneoCurrencies()
     {
diff --git a/Form/Type/AkeneoLocaleType.php b/Form/Type/AkeneoLocaleType.php
index ead1797b..ce69a7e4 100644
--- a/Form/Type/AkeneoLocaleType.php
+++ b/Form/Type/AkeneoLocaleType.php
@@ -12,6 +12,7 @@
 use Symfony\Component\Form\FormBuilderInterface;
 use Symfony\Component\Form\FormEvent;
 use Symfony\Component\Form\FormEvents;
+use Symfony\Component\Intl\Locales;
 use Symfony\Component\OptionsResolver\Exception\AccessException;
 use Symfony\Component\OptionsResolver\OptionsResolver;
 use Symfony\Component\Validator\Constraints\NotBlank;
@@ -47,14 +48,15 @@ public function __construct(LocalizationManager $localizationManager)
      */
     public function buildForm(FormBuilderInterface $builder, array $options)
     {
-        $this->codes = $options['parent_data'];
+        $this->codes = $options['parent_data'] ?? [];
 
         $builder
             ->add(
                 'code',
                 ChoiceType::class,
                 [
-                    'choices' => $this->codes,
+                    'choices' => array_combine($this->codes, $this->codes),
+                    'choice_label' => function ($choice) { return Locales::getName($choice); },
                     'label' => false,
                     'constraints' => [
                         new NotBlank(),
@@ -104,7 +106,8 @@ public function onPreSetData(FormEvent $event)
                 'code',
                 ChoiceType::class,
                 [
-                    'choices' => $this->codes,
+                    'choices' => array_combine($this->codes, $this->codes),
+                    'choice_label' => function ($choice) { return Locales::getName($choice); },
                 ]
             );
     }
@@ -120,7 +123,8 @@ public function onPreSubmit(FormEvent $event)
                     'code',
                     ChoiceType::class,
                     [
-                        'choices' => $this->codes,
+                        'choices' => array_combine($this->codes, $this->codes),
+                        'choice_label' => function ($choice) { return Locales::getName($choice); },
                     ]
                 );
 
diff --git a/Form/Type/AkeneoSettingsType.php b/Form/Type/AkeneoSettingsType.php
index 77f8b508..65c19c41 100644
--- a/Form/Type/AkeneoSettingsType.php
+++ b/Form/Type/AkeneoSettingsType.php
@@ -24,6 +24,7 @@
 use Symfony\Component\Form\FormBuilderInterface;
 use Symfony\Component\Form\FormEvent;
 use Symfony\Component\Form\FormEvents;
+use Symfony\Component\Intl\Locales;
 use Symfony\Component\OptionsResolver\Exception\AccessException;
 use Symfony\Component\OptionsResolver\OptionsResolver;
 use Symfony\Component\Validator\Constraints\NotBlank;
@@ -344,7 +345,7 @@ public function onPreSetData(FormEvent $event)
                 'required'          => true,
                 'label'             => 'oro.akeneo.integration.settings.akeneo_channels.label',
                 'multiple'          => false,
-                'choices'           => $data->getAkeneoChannels(),
+                'choices'           => array_combine($data->getAkeneoChannels() ?? [], $data->getAkeneoChannels() ?? []),
                 'placeholder'       => 'oro.akeneo.integration.settings.akeneo_channels.placeholder',
             ]
         );
@@ -356,7 +357,7 @@ public function onPreSetData(FormEvent $event)
                 'required'          => false,
                 'label'             => 'oro.akeneo.integration.settings.akeneo_currencies.label',
                 'multiple'          => true,
-                'choices'           => $data->getAkeneoCurrencies(),
+                'choices'           => array_combine($data->getAkeneoCurrencies() ?? [], $data->getAkeneoCurrencies() ?? []),
             ]
         );
 
@@ -367,11 +368,12 @@ public function onPreSetData(FormEvent $event)
                 'required'          => false,
                 'label'             => false,
                 'multiple'          => true,
-                'choices'           => $data->getAkeneoLocalesList(),
+                'choices'           => array_combine($data->getAkeneoLocalesList() ?? [], $data->getAkeneoLocalesList() ?? []),
+                'choice_label'      => function ($choice) { return Locales::getName($choice); },
             ]
         );
 
-        $this->codes = $data->getAkeneoLocalesList();
+        $this->codes = $data->getAkeneoLocalesList() ?? [];
         $form->add(
             'akeneoLocales',
             CollectionType::class,
@@ -421,7 +423,7 @@ public function onPreSubmit(FormEvent $event)
                     'required'          => false,
                     'label'             => 'oro.akeneo.integration.settings.akeneo_channels.label',
                     'multiple'          => false,
-                    'choices'           => $channels,
+                    'choices'           => array_combine($channels, $channels),
                 ]
             );
 
@@ -434,12 +436,13 @@ public function onPreSubmit(FormEvent $event)
                     'required'          => false,
                     'label'             => 'oro.akeneo.integration.settings.akeneo_channels.label',
                     'multiple'          => true,
-                    'choices'           => $channels,
+                    'choices'           => array_combine($channels, $channels),
                 ]
             );
 
             $localesList = $this->akeneoTransport->getLocales();
             $transportData['akeneoLocalesList'] = $localesList;
+
             $form->add(
                 'akeneoLocalesList',
                 ChoiceType::class,
@@ -447,7 +450,8 @@ public function onPreSubmit(FormEvent $event)
                     'required'          => false,
                     'label'             => false,
                     'multiple'          => true,
-                    'choices'           => $localesList,
+                    'choices'           => array_combine($localesList, $localesList),
+                    'choice_label'      => function ($choice) { return Locales::getName($choice); },
                 ]
             );
 
@@ -474,7 +478,7 @@ public function onPreSubmit(FormEvent $event)
                     'required'          => false,
                     'label'             => 'oro.akeneo.integration.settings.akeneo_currencies.label',
                     'multiple'          => true,
-                    'choices'           => $currencies,
+                    'choices'           => array_combine($currencies, $currencies),
                 ]
             );
 
@@ -487,7 +491,7 @@ public function onPreSubmit(FormEvent $event)
                     'required'          => false,
                     'label'             => 'oro.akeneo.integration.settings.akeneo_currencies.label',
                     'multiple'          => true,
-                    'choices'           => $currencies,
+                    'choices'           => array_combine($currencies, $currencies),
                 ]
             );
 
diff --git a/Integration/AkeneoTransport.php b/Integration/AkeneoTransport.php
index a6f87e31..982b80fa 100644
--- a/Integration/AkeneoTransport.php
+++ b/Integration/AkeneoTransport.php
@@ -17,7 +17,7 @@
 use Oro\Bundle\IntegrationBundle\Entity\Transport;
 use Psr\Log\LoggerAwareTrait;
 use Psr\Log\LoggerInterface;
-use Symfony\Component\Intl\Languages;
+use Symfony\Component\Intl\Locales;
 
 class AkeneoTransport implements AkeneoTransportInterface
 {
@@ -131,7 +131,7 @@ public function getLocales()
                 continue;
             }
 
-            $localeName = Languages::getName($locale['code']);
+            $localeName = Locales::getName($locale['code']);
             $locales[$localeName ?: $locale['code']] = $locale['code'];
         }
 

From b39ff48a51fade237d835866cdc7e78ccc36de37 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Thu, 10 Mar 2022 01:26:02 +0200
Subject: [PATCH 09/14] AKM-33: OroCommerce 5 LTS compatibility - fix images
 import

---
 .../ImportExportTagsSubscriberDecorator.php   | 21 +++++++++----------
 .../Processor/ProductImageImportProcessor.php |  2 +-
 Resources/config/batch_jobs.yml               | 14 ++++++-------
 Resources/config/services.yml                 | 19 +++++++----------
 4 files changed, 26 insertions(+), 30 deletions(-)

diff --git a/EventListener/ImportExportTagsSubscriberDecorator.php b/EventListener/ImportExportTagsSubscriberDecorator.php
index 63c25167..c8e42c74 100644
--- a/EventListener/ImportExportTagsSubscriberDecorator.php
+++ b/EventListener/ImportExportTagsSubscriberDecorator.php
@@ -13,24 +13,30 @@
 use Oro\Bundle\TagBundle\EventListener\ImportExportTagsSubscriber;
 use Oro\Bundle\TagBundle\Manager\TagImportManager;
 use Symfony\Component\EventDispatcher\EventSubscriberInterface;
-use Symfony\Contracts\Service\ServiceSubscriberInterface;
 
 /**
  * Tags lazy processing
  */
 class ImportExportTagsSubscriberDecorator implements
     AdditionalOptionalListenerInterface,
-    EventSubscriberInterface,
-    ServiceSubscriberInterface
+    EventSubscriberInterface
 {
     use AdditionalOptionalListenerTrait;
 
     /** @var ImportExportTagsSubscriber */
     protected $innerSubscriber;
 
-    public function __construct(ImportExportTagsSubscriber $innerSubscriber)
+    public function __construct(ImportExportTagsSubscriber $innerSubscriber, TagImportManager $tagImportManager)
     {
         $this->innerSubscriber = $innerSubscriber;
+
+        \Closure::bind(
+            function (TagImportManager $tagImportManager) {
+                $this->tagImportManager = $tagImportManager;
+            },
+            $this->innerSubscriber,
+            $this->innerSubscriber
+        )($tagImportManager);
     }
 
     /**
@@ -121,11 +127,4 @@ public function addTagsIntoFixtures(LoadTemplateFixturesEvent $event): void
 
         $this->innerSubscriber->addTagsIntoFixtures($event);
     }
-
-    public static function getSubscribedServices()
-    {
-        return [
-            'oro_tag.tag_import.manager' => TagImportManager::class,
-        ];
-    }
 }
diff --git a/ImportExport/Processor/ProductImageImportProcessor.php b/ImportExport/Processor/ProductImageImportProcessor.php
index 7e54832d..b0caac59 100644
--- a/ImportExport/Processor/ProductImageImportProcessor.php
+++ b/ImportExport/Processor/ProductImageImportProcessor.php
@@ -38,7 +38,7 @@ public function process($items)
             $this->context->setValue('itemData', $image);
 
             /** @var ProductImage $object */
-            $object = $this->serializer->deserialize(
+            $object = $this->serializer->denormalize(
                 $image,
                 $this->getEntityName(),
                 '',
diff --git a/Resources/config/batch_jobs.yml b/Resources/config/batch_jobs.yml
index 3a42d5b7..75fc8a5b 100644
--- a/Resources/config/batch_jobs.yml
+++ b/Resources/config/batch_jobs.yml
@@ -78,13 +78,13 @@ connector:
             reader: oro_akeneo.importexport.reader.product
             processor: oro_akeneo.importexport.processor.product
             writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
-#        import_images:
-#          title: import
-#          class: Oro\Bundle\AkeneoBundle\ImportExport\Step\ItemStep
-#          services:
-#            reader: oro_akeneo.importexport.reader.product_image
-#            processor: oro_akeneo.importexport.processor.product_image
-#            writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
+        import_images:
+          title: import
+          class: Oro\Bundle\AkeneoBundle\ImportExport\Step\ItemStep
+          services:
+            reader: oro_akeneo.importexport.reader.product_image
+            processor: oro_akeneo.importexport.processor.product_image
+            writer: oro_akeneo.importexport.writer.cumulative.persistent_batch_writer
         import_prices:
           title: import
           class: Oro\Bundle\BatchBundle\Step\ItemStep
diff --git a/Resources/config/services.yml b/Resources/config/services.yml
index 23b1b635..88961802 100644
--- a/Resources/config/services.yml
+++ b/Resources/config/services.yml
@@ -141,17 +141,14 @@ services:
     arguments:
       - '@oro_akeneo.event_listener.doctrine_tag.decorator.inner'
 
-#  oro_akeneo.event_listener.import_export_tags_subscriber.decorator:
-#    class: Oro\Bundle\AkeneoBundle\EventListener\ImportExportTagsSubscriberDecorator
-#    decorates: oro_tag.event_listener.import_export_tags_subscriber
-#    decoration_priority: -255
-#    public: true
-#    arguments:
-#      - '@oro_akeneo.event_listener.import_export_tags_subscriber.decorator.inner'
-#
-#  oro_akeneo.tag_import.manager:
-#    alias: oro_tag.tag_import.manager
-#    public: true
+  oro_akeneo.event_listener.import_export_tags_subscriber.decorator:
+    class: Oro\Bundle\AkeneoBundle\EventListener\ImportExportTagsSubscriberDecorator
+    decorates: oro_tag.event_listener.import_export_tags_subscriber
+    decoration_priority: -255
+    public: true
+    arguments:
+      - '@oro_akeneo.event_listener.import_export_tags_subscriber.decorator.inner'
+      - '@oro_tag.tag_import.manager'
 
   oro_akeneo.event_listener.load_class_metadata:
     class: Oro\Bundle\AkeneoBundle\EventListener\LoadClassMetadataListener

From 5747619840be95d3ba98ff6aea22f9841c82b702 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <1804871+dxops@users.noreply.github.com>
Date: Thu, 10 Mar 2022 01:29:33 +0200
Subject: [PATCH 10/14] Allow 5.0 builds

---
 .github/workflows/php.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml
index a3b87ee7..299a7909 100644
--- a/.github/workflows/php.yml
+++ b/.github/workflows/php.yml
@@ -2,7 +2,7 @@ name: PHP Composer
 
 on:
   push:
-    branches: [ master 4.2 4.1 3.1 1.6 ]
+    branches: [ master 5.0 4.2 4.1 3.1 1.6 ]
   pull_request:
 
 jobs:

From 54f69b758a82a47e0c1f9988d0e3395735267b2b Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <1804871+dxops@users.noreply.github.com>
Date: Wed, 13 Apr 2022 21:17:27 +0300
Subject: [PATCH 11/14] AKM-34: Product filter breaks configurable products
 (#65)

AKM-31: Split products and product models filters
- do not allow duplicate jobs
---
 Controller/ValidateConnectionController.php   |   2 +-
 Entity/AkeneoSettings.php                     |  26 ++
 Form/Extension/ChannelTypeExtension.php       |   2 +
 Form/Type/AkeneoSettingsType.php              |   9 +
 ImportExport/Processor/AsyncProcessor.php     |  43 ----
 ImportExport/Writer/AsyncWriter.php           |  99 ++++----
 .../Writer/ConfigurableAsyncWriter.php        | 240 ++++++++++++++++++
 Integration/AkeneoTransport.php               |  74 ++++--
 Integration/AkeneoTransportInterface.php      |   4 +
 .../ConfigurableProductConnector.php          |  72 ++++++
 Integration/Connector/ProductConnector.php    |   5 -
 .../Iterator/ConfigurableProductIterator.php  |  38 +++
 Job/Context/SimpleContextAggregator.php       |   4 +-
 .../Schema/OroAkeneoBundleInstaller.php       |   3 +-
 .../Schema/v1_16/OroAkeneoMigration.php       |  19 ++
 Resources/config/batch_jobs.yml               |  16 +-
 Resources/config/importexport.yml             |  32 ++-
 Resources/translations/messages.en.yml        |   5 +
 Resources/views/Form/fields.html.twig         |  11 +
 .../UniqueProductVariantLinksValidator.php    |  17 +-
 20 files changed, 591 insertions(+), 130 deletions(-)
 create mode 100644 ImportExport/Writer/ConfigurableAsyncWriter.php
 create mode 100644 Integration/Connector/ConfigurableProductConnector.php
 create mode 100644 Integration/Iterator/ConfigurableProductIterator.php
 create mode 100644 Migrations/Schema/v1_16/OroAkeneoMigration.php

diff --git a/Controller/ValidateConnectionController.php b/Controller/ValidateConnectionController.php
index 7cb5c322..ab931f96 100644
--- a/Controller/ValidateConnectionController.php
+++ b/Controller/ValidateConnectionController.php
@@ -113,7 +113,7 @@ public function validateConnectionAction(Request $request, Channel $channel = nu
                 'akeneoLocales' => $akeneoLocales,
                 'success' => $success,
                 'message' => $message,
-                'currencyList' => $this->currencyProvider->getCurrencies(),
+                'currencyList' => $this->currencyProvider->getCurrencyList(),
             ]
         );
     }
diff --git a/Entity/AkeneoSettings.php b/Entity/AkeneoSettings.php
index 3f7f6af9..7e1d61a9 100644
--- a/Entity/AkeneoSettings.php
+++ b/Entity/AkeneoSettings.php
@@ -89,6 +89,12 @@ class AkeneoSettings extends Transport
      * @ORM\Column(name="akeneo_product_filter", type="text", nullable=true)
      */
     protected $productFilter;
+    /**
+     * @var string
+     *
+     * @ORM\Column(name="akeneo_conf_product_filter", type="text", nullable=true)
+     */
+    protected $configurableProductFilter;
     /**
      * @var string
      *
@@ -275,6 +281,26 @@ public function setProductFilter($productFilter)
         return $this;
     }
 
+    /**
+     * @return string
+     */
+    public function getConfigurableProductFilter()
+    {
+        return $this->configurableProductFilter;
+    }
+
+    /**
+     * @param string $configurableProductFilter
+     *
+     * @return self
+     */
+    public function setConfigurableProductFilter($configurableProductFilter)
+    {
+        $this->configurableProductFilter = $configurableProductFilter;
+
+        return $this;
+    }
+
     /**
      * @return ParameterBag
      */
diff --git a/Form/Extension/ChannelTypeExtension.php b/Form/Extension/ChannelTypeExtension.php
index 86ec2363..22912205 100644
--- a/Form/Extension/ChannelTypeExtension.php
+++ b/Form/Extension/ChannelTypeExtension.php
@@ -15,10 +15,12 @@ class ChannelTypeExtension extends AbstractTypeExtension
      * @var array
      */
     protected $connectorsOrder = [
+        'brand',
         'category',
         'attribute',
         'attribute_family',
         'product',
+        'configurable_product',
     ];
 
     /**
diff --git a/Form/Type/AkeneoSettingsType.php b/Form/Type/AkeneoSettingsType.php
index 65c19c41..d2725f78 100644
--- a/Form/Type/AkeneoSettingsType.php
+++ b/Form/Type/AkeneoSettingsType.php
@@ -249,6 +249,15 @@ public function buildForm(FormBuilderInterface $builder, array $options)
                     ],
                 ]
             )
+            ->add(
+                'configurableProductFilter',
+                TextareaType::class,
+                [
+                    'required'    => false,
+                    'label'       => 'oro.akeneo.integration.settings.akeneo_configurable_product_filter.label',
+                    'constraints' => [new JsonConstraint()],
+                ]
+            )
             ->add(
                 'priceList',
                 PriceListSelectType::class,
diff --git a/ImportExport/Processor/AsyncProcessor.php b/ImportExport/Processor/AsyncProcessor.php
index 1c73739d..31fab7b7 100644
--- a/ImportExport/Processor/AsyncProcessor.php
+++ b/ImportExport/Processor/AsyncProcessor.php
@@ -6,51 +6,8 @@
 
 class AsyncProcessor implements ProcessorInterface
 {
-    use CacheProviderAwareProcessor;
-
-    /** @var array */
-    private $variants = [];
-
     public function process($item)
     {
-        $this->updateVariants($item);
-
         return $item;
     }
-
-    private function updateVariants(array &$item)
-    {
-        $sku = $item['sku'];
-
-        if (!empty($item['family_variant'])) {
-            if (isset($item['parent'], $this->variants[$sku])) {
-                $parent = $item['parent'];
-                foreach (array_keys($this->variants[$sku]) as $sku) {
-                    $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
-                }
-            }
-
-            return;
-        }
-
-        if (empty($item['parent'])) {
-            return;
-        }
-
-        $parent = $item['parent'];
-
-        $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
-    }
-
-    public function initialize()
-    {
-        $this->variants = [];
-        $this->cacheProvider->delete('product_variants');
-    }
-
-    public function flush()
-    {
-        $this->cacheProvider->save('product_variants', $this->variants);
-        $this->variants = [];
-    }
 }
diff --git a/ImportExport/Writer/AsyncWriter.php b/ImportExport/Writer/AsyncWriter.php
index bf88761a..5bb56f57 100644
--- a/ImportExport/Writer/AsyncWriter.php
+++ b/ImportExport/Writer/AsyncWriter.php
@@ -5,7 +5,7 @@
 use Doctrine\DBAL\Platforms\MySqlPlatform;
 use Doctrine\DBAL\Types\Types;
 use Oro\Bundle\AkeneoBundle\Async\Topics;
-use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
+use Oro\Bundle\AkeneoBundle\EventListener\AdditionalOptionalListenerManager;
 use Oro\Bundle\BatchBundle\Entity\StepExecution;
 use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
 use Oro\Bundle\BatchBundle\Item\Support\ClosableInterface;
@@ -14,6 +14,7 @@
 use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
 use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
 use Oro\Bundle\MessageQueueBundle\Entity\Job;
+use Oro\Bundle\PlatformBundle\Manager\OptionalListenerManager;
 use Oro\Component\MessageQueue\Client\Message;
 use Oro\Component\MessageQueue\Client\MessagePriority;
 use Oro\Component\MessageQueue\Client\MessageProducerInterface;
@@ -23,10 +24,6 @@ class AsyncWriter implements
     ClosableInterface,
     StepExecutionAwareInterface
 {
-    use CacheProviderTrait;
-
-    private const VARIANTS_BATCH_SIZE = 25;
-
     /** @var MessageProducerInterface * */
     private $messageProducer;
 
@@ -39,17 +36,30 @@ class AsyncWriter implements
     /** @var DoctrineHelper */
     private $doctrineHelper;
 
+    /** @var OptionalListenerManager */
+    private $optionalListenerManager;
+
+    /** @var AdditionalOptionalListenerManager */
+    private $additionalOptionalListenerManager;
+
     public function __construct(
         MessageProducerInterface $messageProducer,
-        DoctrineHelper $doctrineHelper
+        DoctrineHelper $doctrineHelper,
+        OptionalListenerManager $optionalListenerManager,
+        AdditionalOptionalListenerManager $additionalOptionalListenerManager
     ) {
         $this->messageProducer = $messageProducer;
         $this->doctrineHelper = $doctrineHelper;
+        $this->optionalListenerManager = $optionalListenerManager;
+        $this->additionalOptionalListenerManager = $additionalOptionalListenerManager;
     }
 
     public function initialize()
     {
         $this->size = 0;
+
+        $this->additionalOptionalListenerManager->disableListeners();
+        $this->optionalListenerManager->disableListeners($this->optionalListenerManager->getListeners());
     }
 
     public function write(array $items)
@@ -67,52 +77,30 @@ public function write(array $items)
         $this->stepExecution->setWriteCount($this->size);
 
         $jobId = $this->insertJob($jobName);
-        $this->createFieldsChanges($jobId, $items, 'items');
-        $this->sendMessage($channelId, $jobId, true);
-    }
-
-    public function flush()
-    {
-        $this->size = 0;
-
-        $variants = $this->cacheProvider->fetch('product_variants') ?? [];
-        if (!$variants) {
-            return;
-        }
-
-        $channelId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('channel');
-
-        $chunks = array_chunk($variants, self::VARIANTS_BATCH_SIZE, true);
-
-        foreach ($chunks as $key => $chunk) {
-            $jobName = sprintf(
-                'oro_integration:sync_integration:%s:variants:%s-%s',
-                $channelId,
-                self::VARIANTS_BATCH_SIZE * $key + 1,
-                self::VARIANTS_BATCH_SIZE * $key + count($chunk)
-            );
-
-            $jobId = $this->insertJob($jobName);
-            $this->createFieldsChanges($jobId, $chunk, 'variants');
-            $this->sendMessage($channelId, $jobId);
+        if ($jobId && $this->createFieldsChanges($jobId, $items, 'items')) {
+            $this->sendMessage($channelId, $jobId, true);
         }
     }
 
-    private function createFieldsChanges(int $jobId, array &$data, string $key): void
+    private function createFieldsChanges(int $jobId, array &$data, string $key): bool
     {
         $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
         $fieldsChanges = $em
             ->getRepository(FieldsChanges::class)
             ->findOneBy(['entityId' => $jobId, 'entityClass' => Job::class]);
-        if (!$fieldsChanges) {
-            $fieldsChanges = new FieldsChanges([]);
-            $fieldsChanges->setEntityClass(Job::class);
-            $fieldsChanges->setEntityId($jobId);
+        if ($fieldsChanges) {
+            return false;
         }
+
+        $fieldsChanges = new FieldsChanges([]);
+        $fieldsChanges->setEntityClass(Job::class);
+        $fieldsChanges->setEntityId($jobId);
         $fieldsChanges->setChangedFields([$key => $data]);
         $em->persist($fieldsChanges);
         $em->flush($fieldsChanges);
         $em->clear(FieldsChanges::class);
+
+        return true;
     }
 
     private function sendMessage(int $channelId, int $jobId, bool $incrementedRead = false): void
@@ -143,12 +131,15 @@ private function getRootJob(): ?int
             throw new \InvalidArgumentException('Root job id is empty');
         }
 
-        return $rootJobId;
+        return (int)$rootJobId;
     }
 
     public function close()
     {
         $this->size = 0;
+
+        $this->optionalListenerManager->enableListeners($this->optionalListenerManager->getListeners());
+        $this->additionalOptionalListenerManager->enableListeners();
     }
 
     public function setStepExecution(StepExecution $stepExecution)
@@ -159,12 +150,34 @@ public function setStepExecution(StepExecution $stepExecution)
     private function insertJob(string $jobName): ?int
     {
         $em = $this->doctrineHelper->getEntityManager(Job::class);
-        $tableName = $em->getClassMetadata(Job::class)->getTableName();
         $connection = $em->getConnection();
+        $rootJobId = $this->getRootJob();
+
+        $hasRootJob = $connection
+            ->executeStatement(
+                'SELECT 1 FROM oro_message_queue_job WHERE id = :id LIMIT 1;',
+                ['id' => $rootJobId],
+                ['id' => Types::INTEGER]
+            );
+
+        if (!$hasRootJob) {
+            throw new \InvalidArgumentException(sprintf('Root job "%d" missing', $rootJobId));
+        }
+
+        $childJob = $connection
+            ->executeStatement(
+                'SELECT id FROM oro_message_queue_job WHERE root_job_id = :rootJob and name = :name LIMIT 1;',
+                ['rootJob' => $rootJobId, 'name' => $jobName],
+                ['rootJob' => Types::INTEGER, 'name' => Types::STRING]
+            );
+
+        if ($childJob) {
+            return $childJob;
+        }
 
         $qb = $connection->createQueryBuilder();
         $qb
-            ->insert($tableName)
+            ->insert('oro_message_queue_job')
             ->values([
                 'name' => ':name',
                 'status' => ':status',
@@ -178,7 +191,7 @@ private function insertJob(string $jobName): ?int
                 'interrupted' => false,
                 'unique' => false,
                 'createdAt' => new \DateTime(),
-                'rootJob' => $this->getRootJob(),
+                'rootJob' => $rootJobId,
             ], [
                 'name' => Types::STRING,
                 'status' => Types::STRING,
diff --git a/ImportExport/Writer/ConfigurableAsyncWriter.php b/ImportExport/Writer/ConfigurableAsyncWriter.php
new file mode 100644
index 00000000..c1e55588
--- /dev/null
+++ b/ImportExport/Writer/ConfigurableAsyncWriter.php
@@ -0,0 +1,240 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\ImportExport\Writer;
+
+use Doctrine\DBAL\Platforms\MySqlPlatform;
+use Doctrine\DBAL\Types\Types;
+use Oro\Bundle\AkeneoBundle\Async\Topics;
+use Oro\Bundle\AkeneoBundle\EventListener\AdditionalOptionalListenerManager;
+use Oro\Bundle\BatchBundle\Entity\StepExecution;
+use Oro\Bundle\BatchBundle\Item\ItemWriterInterface;
+use Oro\Bundle\BatchBundle\Step\StepExecutionAwareInterface;
+use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
+use Oro\Bundle\IntegrationBundle\Entity\FieldsChanges;
+use Oro\Bundle\MessageQueueBundle\Client\BufferedMessageProducer;
+use Oro\Bundle\MessageQueueBundle\Entity\Job;
+use Oro\Bundle\PlatformBundle\Manager\OptionalListenerManager;
+use Oro\Component\MessageQueue\Client\Message;
+use Oro\Component\MessageQueue\Client\MessagePriority;
+use Oro\Component\MessageQueue\Client\MessageProducerInterface;
+
+class ConfigurableAsyncWriter implements
+    ItemWriterInterface,
+    StepExecutionAwareInterface
+{
+    private const VARIANTS_BATCH_SIZE = 25;
+
+    /** @var MessageProducerInterface * */
+    private $messageProducer;
+
+    /** @var StepExecution */
+    private $stepExecution;
+
+    /** @var DoctrineHelper */
+    private $doctrineHelper;
+
+    /** @var OptionalListenerManager */
+    private $optionalListenerManager;
+
+    /** @var AdditionalOptionalListenerManager */
+    private $additionalOptionalListenerManager;
+
+    private $variants = [];
+
+    public function __construct(
+        MessageProducerInterface $messageProducer,
+        DoctrineHelper $doctrineHelper,
+        OptionalListenerManager $optionalListenerManager,
+        AdditionalOptionalListenerManager $additionalOptionalListenerManager
+    ) {
+        $this->messageProducer = $messageProducer;
+        $this->doctrineHelper = $doctrineHelper;
+        $this->optionalListenerManager = $optionalListenerManager;
+        $this->additionalOptionalListenerManager = $additionalOptionalListenerManager;
+    }
+
+    public function initialize()
+    {
+        $this->variants = [];
+
+        $this->additionalOptionalListenerManager->disableListeners();
+        $this->optionalListenerManager->disableListeners($this->optionalListenerManager->getListeners());
+    }
+
+    public function write(array $items)
+    {
+        foreach ($items as $item) {
+            $sku = $item['sku'];
+
+            if (!empty($item['family_variant'])) {
+                if (isset($item['parent'], $this->variants[$sku])) {
+                    $parent = $item['parent'];
+                    foreach (array_keys($this->variants[$sku]) as $sku) {
+                        $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
+                    }
+                }
+
+                return;
+            }
+
+            if (empty($item['parent'])) {
+                return;
+            }
+
+            $parent = $item['parent'];
+
+            $this->variants[$parent][$sku] = ['parent' => $parent, 'variant' => $sku];
+        }
+    }
+
+    public function close()
+    {
+        $this->variants = [];
+
+        $this->optionalListenerManager->enableListeners($this->optionalListenerManager->getListeners());
+        $this->additionalOptionalListenerManager->enableListeners();
+    }
+
+    public function flush()
+    {
+        $channelId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('channel');
+
+        $chunks = array_chunk($this->variants, self::VARIANTS_BATCH_SIZE, true);
+
+        foreach ($chunks as $key => $chunk) {
+            $jobName = sprintf(
+                'oro_integration:sync_integration:%s:variants:%s-%s',
+                $channelId,
+                self::VARIANTS_BATCH_SIZE * $key + 1,
+                self::VARIANTS_BATCH_SIZE * $key + count($chunk)
+            );
+
+            $jobId = $this->insertJob($jobName);
+            if ($jobId && $this->createFieldsChanges($jobId, $chunk, 'variants')) {
+                $this->sendMessage($channelId, $jobId);
+            }
+        }
+    }
+
+    private function createFieldsChanges(int $jobId, array &$data, string $key): bool
+    {
+        $em = $this->doctrineHelper->getEntityManager(FieldsChanges::class);
+        $fieldsChanges = $em
+            ->getRepository(FieldsChanges::class)
+            ->findOneBy(['entityId' => $jobId, 'entityClass' => Job::class]);
+        if ($fieldsChanges) {
+            return false;
+        }
+
+        $fieldsChanges = new FieldsChanges([]);
+        $fieldsChanges->setEntityClass(Job::class);
+        $fieldsChanges->setEntityId($jobId);
+        $fieldsChanges->setChangedFields([$key => $data]);
+        $em->persist($fieldsChanges);
+        $em->flush($fieldsChanges);
+        $em->clear(FieldsChanges::class);
+
+        return true;
+    }
+
+    private function sendMessage(int $channelId, int $jobId, bool $incrementedRead = false): void
+    {
+        $this->messageProducer->send(
+            Topics::IMPORT_PRODUCTS,
+            new Message(
+                [
+                    'integrationId' => $channelId,
+                    'jobId' => $jobId,
+                    'connector' => 'configurable_product',
+                    'connector_parameters' => ['incremented_read' => $incrementedRead],
+                ],
+                MessagePriority::HIGH
+            )
+        );
+
+        if ($this->messageProducer instanceof BufferedMessageProducer
+            && $this->messageProducer->isBufferingEnabled()) {
+            $this->messageProducer->flushBuffer();
+        }
+    }
+
+    private function getRootJob(): ?int
+    {
+        $rootJobId = $this->stepExecution->getJobExecution()->getExecutionContext()->get('rootJobId') ?? null;
+        if (!$rootJobId) {
+            throw new \InvalidArgumentException('Root job id is empty');
+        }
+
+        return (int)$rootJobId;
+    }
+
+    public function setStepExecution(StepExecution $stepExecution)
+    {
+        $this->stepExecution = $stepExecution;
+    }
+
+    private function insertJob(string $jobName): ?int
+    {
+        $em = $this->doctrineHelper->getEntityManager(Job::class);
+        $connection = $em->getConnection();
+        $rootJobId = $this->getRootJob();
+
+        $hasRootJob = $connection
+            ->executeStatement(
+                'SELECT 1 FROM oro_message_queue_job WHERE id = :id LIMIT 1;',
+                ['id' => $rootJobId],
+                ['id' => Types::INTEGER]
+            );
+
+        if (!$hasRootJob) {
+            throw new \InvalidArgumentException(sprintf('Root job "%d" missing', $rootJobId));
+        }
+
+        $childJob = $connection
+            ->executeStatement(
+                'SELECT id FROM oro_message_queue_job WHERE root_job_id = :rootJob and name = :name LIMIT 1;',
+                ['rootJob' => $rootJobId, 'name' => $jobName],
+                ['rootJob' => Types::INTEGER, 'name' => Types::STRING]
+            );
+
+        if ($childJob) {
+            return $childJob;
+        }
+
+        $qb = $connection->createQueryBuilder();
+        $qb
+            ->insert('oro_message_queue_job')
+            ->values([
+                'name' => ':name',
+                'status' => ':status',
+                'interrupted' => ':interrupted',
+                'created_at' => ':createdAt',
+                'root_job_id' => ':rootJob',
+            ])
+            ->setParameters([
+                'name' => $jobName,
+                'status' => Job::STATUS_NEW,
+                'interrupted' => false,
+                'unique' => false,
+                'createdAt' => new \DateTime(),
+                'rootJob' => $rootJobId,
+            ], [
+                'name' => Types::STRING,
+                'status' => Types::STRING,
+                'interrupted' => Types::BOOLEAN,
+                'unique' => Types::BOOLEAN,
+                'createdAt' => Types::DATETIME_MUTABLE,
+                'rootJob' => Types::INTEGER,
+            ]);
+
+        if ($connection->getDatabasePlatform() instanceof MySqlPlatform) {
+            $qb->setValue('`unique`', ':unique');
+        } else {
+            $qb->setValue('"unique"', ':unique');
+        }
+
+        $qb->execute();
+
+        return $connection->lastInsertId();
+    }
+}
diff --git a/Integration/AkeneoTransport.php b/Integration/AkeneoTransport.php
index 982b80fa..1aaefa37 100644
--- a/Integration/AkeneoTransport.php
+++ b/Integration/AkeneoTransport.php
@@ -10,6 +10,7 @@
 use Oro\Bundle\AkeneoBundle\Integration\Iterator\AttributeFamilyIterator;
 use Oro\Bundle\AkeneoBundle\Integration\Iterator\AttributeIterator;
 use Oro\Bundle\AkeneoBundle\Integration\Iterator\BrandIterator;
+use Oro\Bundle\AkeneoBundle\Integration\Iterator\ConfigurableProductIterator;
 use Oro\Bundle\AkeneoBundle\Integration\Iterator\ProductIterator;
 use Oro\Bundle\AkeneoBundle\Settings\DataProvider\SyncProductsDataProvider;
 use Oro\Bundle\CurrencyBundle\Provider\CurrencyProviderInterface;
@@ -209,14 +210,14 @@ public function getProducts(int $pageSize)
         $this->initAttributesList();
         $this->initMeasureFamilies();
 
-        $searchFilters = $this->akeneoSearchBuilder->getFilters($this->transportEntity->getProductFilter());
+        $queryParams = [
+            'scope' => $this->transportEntity->getAkeneoActiveChannel(),
+            'search' => $this->akeneoSearchBuilder->getFilters($this->transportEntity->getProductFilter()),
+        ];
 
         if ($this->transportEntity->getSyncProducts() === SyncProductsDataProvider::PUBLISHED) {
             return new ProductIterator(
-                $this->client->getPublishedProductApi()->all(
-                    $pageSize,
-                    ['search' => $searchFilters, 'scope' => $this->transportEntity->getAkeneoActiveChannel()]
-                ),
+                $this->client->getPublishedProductApi()->all($pageSize, $queryParams),
                 $this->client,
                 $this->logger,
                 $this->attributes,
@@ -227,10 +228,7 @@ public function getProducts(int $pageSize)
         }
 
         return new ProductIterator(
-            $this->client->getProductApi()->all(
-                $pageSize,
-                ['search' => $searchFilters, 'scope' => $this->transportEntity->getAkeneoActiveChannel()]
-            ),
+            $this->client->getProductApi()->all($pageSize, $queryParams),
             $this->client,
             $this->logger,
             $this->attributes,
@@ -240,6 +238,33 @@ public function getProducts(int $pageSize)
         );
     }
 
+    public function getProductsList(int $pageSize): iterable
+    {
+        $this->initAttributesList();
+
+        $queryParams = [
+            'scope' => $this->transportEntity->getAkeneoActiveChannel(),
+            'search' => $this->akeneoSearchBuilder->getFilters($this->transportEntity->getProductFilter()),
+            'attributes' => key($this->attributes),
+        ];
+
+        if ($this->transportEntity->getSyncProducts() === SyncProductsDataProvider::PUBLISHED) {
+            return new ConfigurableProductIterator(
+                $this->client->getPublishedProductApi()->all($pageSize, $queryParams),
+                $this->client,
+                $this->logger,
+                $this->getAttributeMapping()
+            );
+        }
+
+        return new ConfigurableProductIterator(
+            $this->client->getProductApi()->all($pageSize, $queryParams),
+            $this->client,
+            $this->logger,
+            $this->getAttributeMapping()
+        );
+    }
+
     /**
      * @return \Iterator
      */
@@ -249,16 +274,13 @@ public function getProductModels(int $pageSize)
         $this->initFamilyVariants();
         $this->initMeasureFamilies();
 
-        $searchFilters = $this->akeneoSearchBuilder->getFilters($this->transportEntity->getProductFilter());
-        if (isset($searchFilters['completeness'])) {
-            unset($searchFilters['completeness']);
-        }
+        $queryParams = [
+            'scope' => $this->transportEntity->getAkeneoActiveChannel(),
+            'search' => $this->akeneoSearchBuilder->getFilters($this->transportEntity->getConfigurableProductFilter()),
+        ];
 
         return new ProductIterator(
-            $this->client->getProductModelApi()->all(
-                $pageSize,
-                ['search' => $searchFilters, 'scope' => $this->transportEntity->getAkeneoActiveChannel()]
-            ),
+            $this->client->getProductModelApi()->all($pageSize, $queryParams),
             $this->client,
             $this->logger,
             $this->attributes,
@@ -268,6 +290,24 @@ public function getProductModels(int $pageSize)
         );
     }
 
+    public function getProductModelsList(int $pageSize): iterable
+    {
+        $this->initAttributesList();
+
+        $queryParams = [
+            'scope' => $this->transportEntity->getAkeneoActiveChannel(),
+            'search' => $this->akeneoSearchBuilder->getFilters($this->transportEntity->getConfigurableProductFilter()),
+            'attributes' => key($this->attributes),
+        ];
+
+        return new ConfigurableProductIterator(
+            $this->client->getProductModelApi()->all($pageSize, $queryParams),
+            $this->client,
+            $this->logger,
+            $this->getAttributeMapping()
+        );
+    }
+
     /**
      * {@inheritdoc}
      */
diff --git a/Integration/AkeneoTransportInterface.php b/Integration/AkeneoTransportInterface.php
index 25ef9902..2f26b9b9 100644
--- a/Integration/AkeneoTransportInterface.php
+++ b/Integration/AkeneoTransportInterface.php
@@ -51,6 +51,10 @@ public function getProducts(int $pageSize);
      */
     public function getProductModels(int $pageSize);
 
+    public function getProductsList(int $pageSize): iterable;
+
+    public function getProductModelsList(int $pageSize): iterable;
+
     /**
      * @return \Iterator
      */
diff --git a/Integration/Connector/ConfigurableProductConnector.php b/Integration/Connector/ConfigurableProductConnector.php
new file mode 100644
index 00000000..ecb0ce15
--- /dev/null
+++ b/Integration/Connector/ConfigurableProductConnector.php
@@ -0,0 +1,72 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Integration\Connector;
+
+use Oro\Bundle\AkeneoBundle\Integration\AkeneoTransport;
+use Oro\Bundle\AkeneoBundle\Placeholder\SchemaUpdateFilter;
+use Oro\Bundle\AkeneoBundle\Tools\CacheProviderTrait;
+use Oro\Bundle\IntegrationBundle\Entity\Channel;
+use Oro\Bundle\IntegrationBundle\Provider\AbstractConnector;
+use Oro\Bundle\IntegrationBundle\Provider\AllowedConnectorInterface;
+use Oro\Bundle\IntegrationBundle\Provider\ConnectorInterface;
+use Oro\Bundle\ProductBundle\Entity\Product;
+
+/**
+ * Integration configurable product connector.
+ */
+class ConfigurableProductConnector extends AbstractConnector implements ConnectorInterface, AllowedConnectorInterface
+{
+    use CacheProviderTrait;
+
+    const PAGE_SIZE = 100;
+
+    /** @var AkeneoTransport */
+    protected $transport;
+
+    /** @var SchemaUpdateFilter */
+    protected $schemaUpdateFilter;
+
+    public function getLabel(): string
+    {
+        return 'oro.akeneo.connector.configurable_product.label';
+    }
+
+    public function getImportEntityFQCN()
+    {
+        return Product::class;
+    }
+
+    public function getImportJobName()
+    {
+        return 'akeneo_configurable_product_import';
+    }
+
+    public function getType()
+    {
+        return 'configurable_product';
+    }
+
+    public function setSchemaUpdateFilter(SchemaUpdateFilter $schemaUpdateFilter): void
+    {
+        $this->schemaUpdateFilter = $schemaUpdateFilter;
+    }
+
+    public function isAllowed(Channel $integration, array $processedConnectorsStatuses): bool
+    {
+        return $this->schemaUpdateFilter->isApplicable($integration, Product::class) === false;
+    }
+
+    protected function getConnectorSource()
+    {
+        $variants = $this->cacheProvider->fetch('akeneo')['variants'] ?? [];
+        if ($variants) {
+            return new \ArrayIterator();
+        }
+
+        $iterator = new \AppendIterator();
+        $iterator->append($this->transport->getProductsList(self::PAGE_SIZE));
+        $iterator->append($this->transport->getProductModelsList(self::PAGE_SIZE));
+
+        return $iterator;
+    }
+}
diff --git a/Integration/Connector/ProductConnector.php b/Integration/Connector/ProductConnector.php
index 95bd70f4..a5ae32e8 100644
--- a/Integration/Connector/ProductConnector.php
+++ b/Integration/Connector/ProductConnector.php
@@ -81,11 +81,6 @@ protected function getConnectorSource()
             return new \ArrayIterator();
         }
 
-        $variants = $this->cacheProvider->fetch('akeneo')['variants'] ?? [];
-        if ($variants) {
-            return new \ArrayIterator();
-        }
-
         $iterator = new \AppendIterator();
         $iterator->append($this->transport->getProducts(self::PAGE_SIZE));
         $iterator->append($this->transport->getProductModels(self::PAGE_SIZE));
diff --git a/Integration/Iterator/ConfigurableProductIterator.php b/Integration/Iterator/ConfigurableProductIterator.php
new file mode 100644
index 00000000..4a533506
--- /dev/null
+++ b/Integration/Iterator/ConfigurableProductIterator.php
@@ -0,0 +1,38 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Integration\Iterator;
+
+use Akeneo\Pim\ApiClient\AkeneoPimClientInterface;
+use Akeneo\Pim\ApiClient\Pagination\ResourceCursorInterface;
+use Psr\Log\LoggerInterface;
+
+class ConfigurableProductIterator extends AbstractIterator
+{
+    private $attributeMapping = [];
+
+    public function __construct(
+        ResourceCursorInterface $resourceCursor,
+        AkeneoPimClientInterface $client,
+        LoggerInterface $logger,
+        array $attributeMapping = []
+    ) {
+        parent::__construct($resourceCursor, $client, $logger);
+
+        $this->attributeMapping = $attributeMapping;
+    }
+
+    public function doCurrent()
+    {
+        $item = $this->resourceCursor->current();
+
+        $sku = $item['identifier'] ?? $item['code'];
+
+        if (array_key_exists('sku', $this->attributeMapping)) {
+            if (!empty($item['values'][$this->attributeMapping['sku']][0]['data'])) {
+                $sku = $item['values'][$this->attributeMapping['sku']][0]['data'];
+            }
+        }
+
+        return ['sku' => (string)$sku, 'parent' => $item['parent'] ?? null, 'family_variant' => $item['family_variant'] ?? null];
+    }
+}
diff --git a/Job/Context/SimpleContextAggregator.php b/Job/Context/SimpleContextAggregator.php
index c98e8a7e..869bdc7f 100644
--- a/Job/Context/SimpleContextAggregator.php
+++ b/Job/Context/SimpleContextAggregator.php
@@ -25,9 +25,9 @@ public function getAggregatedContext(JobExecution $jobExecution)
                     $context,
                     $this->contextRegistry->getByStepExecution($stepExecution)
                 );
-                //CUSTOMIZATION START
+                // CUSTOMIZATION START
                 $context->addErrors($stepExecution->getErrors());
-                //CUSTOMIZATION END
+                // CUSTOMIZATION END
             }
         }
 
diff --git a/Migrations/Schema/OroAkeneoBundleInstaller.php b/Migrations/Schema/OroAkeneoBundleInstaller.php
index 54aa4f57..62f6f206 100644
--- a/Migrations/Schema/OroAkeneoBundleInstaller.php
+++ b/Migrations/Schema/OroAkeneoBundleInstaller.php
@@ -47,7 +47,7 @@ class OroAkeneoBundleInstaller implements Installation, ExtendExtensionAwareInte
      */
     public function getMigrationVersion()
     {
-        return 'v1_15';
+        return 'v1_16';
     }
 
     /**
@@ -113,6 +113,7 @@ protected function updateIntegrationTransportTable(Schema $schema)
         $table->addColumn('akeneo_active_channel', 'string', ['notnull' => false, 'length' => 255]);
         $table->addColumn('akeneo_acl_voter_enabled', 'boolean', ['notnull' => false]);
         $table->addColumn('akeneo_product_filter', 'text', ['notnull' => false]);
+        $table->addColumn('akeneo_conf_product_filter', 'text', ['notnull' => false]);
         $table->addColumn('akeneo_attributes_list', 'text', ['notnull' => false]);
         $table->addColumn('rootcategory_id', 'integer', ['notnull' => false]);
         $table->addColumn('pricelist_id', 'integer', ['notnull' => false]);
diff --git a/Migrations/Schema/v1_16/OroAkeneoMigration.php b/Migrations/Schema/v1_16/OroAkeneoMigration.php
new file mode 100644
index 00000000..8c1a2368
--- /dev/null
+++ b/Migrations/Schema/v1_16/OroAkeneoMigration.php
@@ -0,0 +1,19 @@
+<?php
+
+namespace Oro\Bundle\AkeneoBundle\Migrations\Schema\v1_16;
+
+use Doctrine\DBAL\Schema\Schema;
+use Oro\Bundle\MigrationBundle\Migration\Migration;
+use Oro\Bundle\MigrationBundle\Migration\QueryBag;
+
+class OroAkeneoMigration implements Migration
+{
+    public function up(Schema $schema, QueryBag $queries)
+    {
+        $table = $schema->getTable('oro_integration_transport');
+        $table->addColumn('akeneo_conf_product_filter', 'text', ['notnull' => false]);
+
+        $queries->addPostQuery('UPDATE oro_integration_transport SET akeneo_conf_product_filter = akeneo_product_filter ' .
+            'WHERE akeneo_product_filter IS NOT NULL;');
+    }
+}
diff --git a/Resources/config/batch_jobs.yml b/Resources/config/batch_jobs.yml
index 75fc8a5b..d6505d30 100644
--- a/Resources/config/batch_jobs.yml
+++ b/Resources/config/batch_jobs.yml
@@ -92,9 +92,23 @@ connector:
             reader: oro_akeneo.importexport.reader.price
             processor: oro_akeneo.importexport.processor.import.product_price
             writer: oro_pricing.importexport.writer.product_price
-        import_variants:
+
+    akeneo_configurable_product_import:
+      title: "Configurable Product import from Akeneo"
+      type:  import
+      steps:
+        api:
           title: import
           class: Oro\Bundle\BatchBundle\Step\ItemStep
+          services:
+            reader: oro_akeneo.integration.connector.configurable_product
+            processor: oro_akeneo.importexport.processor.async
+            writer: oro_akeneo.importexport.writer.configurable_async_product
+          parameters:
+            batch_size: 25
+        import_variants:
+          title: import
+          class: Oro\Bundle\AkeneoBundle\ImportExport\Step\ItemStep
           services:
             reader: oro_akeneo.importexport.reader.product_variant
             processor: oro_akeneo.importexport.processor.import.product_variant
diff --git a/Resources/config/importexport.yml b/Resources/config/importexport.yml
index c6a34739..a7d4d114 100644
--- a/Resources/config/importexport.yml
+++ b/Resources/config/importexport.yml
@@ -106,8 +106,6 @@ services:
   oro_akeneo.importexport.processor.async:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Processor\AsyncProcessor'
     public: true
-    calls:
-      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
   oro_akeneo.importexport.processor.category_parent:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\Processor\CategoryParentProcessor'
@@ -202,6 +200,15 @@ services:
     calls:
       - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
 
+  oro_akeneo.integration.connector.brand:
+    class: 'Oro\Bundle\AkeneoBundle\Integration\Connector\BrandConnector'
+    arguments:
+      - '@oro_importexport.context_registry'
+      - '@oro_integration.logger.strategy'
+      - '@oro_integration.provider.connector_context_mediator'
+    tags:
+      - { name: oro_integration.connector, type: brand, channel_type: oro_akeneo }
+
   oro_akeneo.integration.connector.product:
     class: 'Oro\Bundle\AkeneoBundle\Integration\Connector\ProductConnector'
     arguments:
@@ -214,14 +221,17 @@ services:
     tags:
       - { name: oro_integration.connector, type: product, channel_type: oro_akeneo }
 
-  oro_akeneo.integration.connector.brand:
-    class: 'Oro\Bundle\AkeneoBundle\Integration\Connector\BrandConnector'
+  oro_akeneo.integration.connector.configurable_product:
+    class: 'Oro\Bundle\AkeneoBundle\Integration\Connector\ConfigurableProductConnector'
     arguments:
       - '@oro_importexport.context_registry'
       - '@oro_integration.logger.strategy'
       - '@oro_integration.provider.connector_context_mediator'
+    calls:
+      - [ setSchemaUpdateFilter, [ '@oro_akeneo.placeholder.schema_update_filter' ] ]
+      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
     tags:
-      - { name: oro_integration.connector, type: brand, channel_type: oro_akeneo }
+      - { name: oro_integration.connector, type: configurable_product, channel_type: oro_akeneo }
 
   oro_akeneo.importexport.data_converter.product:
     class: 'Oro\Bundle\AkeneoBundle\ImportExport\DataConverter\ProductDataConverter'
@@ -315,8 +325,16 @@ services:
     arguments:
       - '@oro_message_queue.message_producer'
       - '@oro_entity.doctrine_helper'
-    calls:
-      - [ setCacheProvider, [ '@oro_akeneo.importexport.cache' ] ]
+      - '@oro_platform.optional_listeners.manager'
+      - '@oro_akeneo.event_listener.additional_optional_listeners_manager'
+
+  oro_akeneo.importexport.writer.configurable_async_product:
+    class: 'Oro\Bundle\AkeneoBundle\ImportExport\Writer\ConfigurableAsyncWriter'
+    arguments:
+      - '@oro_message_queue.message_producer'
+      - '@oro_entity.doctrine_helper'
+      - '@oro_platform.optional_listeners.manager'
+      - '@oro_akeneo.event_listener.additional_optional_listeners_manager'
 
   oro_akeneo.importexport.cache:
     parent: oro_cache.array_cache
diff --git a/Resources/translations/messages.en.yml b/Resources/translations/messages.en.yml
index 7773d8ad..26bd0041 100644
--- a/Resources/translations/messages.en.yml
+++ b/Resources/translations/messages.en.yml
@@ -38,6 +38,9 @@ oro:
         akeneo_product_filter:
           label: 'Product Filter'
           tooltip: 'This field enables you to apply filters to sync only the products you want. As this filter is passed via API request, it must be filled in JSON format. Details on the format and filter options available for the products can be found in the <a target="_blank" href="https://api.akeneo.com/documentation/filter.html#filters">Filters section of the Akeneo PIM documentation</a>'
+        akeneo_configurable_product_filter:
+          label: 'Configurable Product Filter'
+          tooltip: 'This field enables you to apply filters to sync only the configurable products you want. As this filter is passed via API request, it must be filled in JSON format. Details on the format and filter options available for the products can be found in the <a target="_blank" href="https://api.akeneo.com/documentation/filter.html#filters">Filters section of the Akeneo PIM documentation</a>'
         akeneo_attribute_list:
           label: 'Attribute Filter'
           tooltip: 'This field enables you to apply filters to sync only the attributes you want. Values must be attribute code, separated with a semi-colon. IMPORTANT: if not defined before to save the integration, all attributes will be imported.'
@@ -84,6 +87,8 @@ oro:
         label: Category connector
       product:
         label: Product connector
+      configurable_product:
+        label: Configurable product connector
       attribute_family:
         label: Attribute family connector
       attribute:
diff --git a/Resources/views/Form/fields.html.twig b/Resources/views/Form/fields.html.twig
index 5a09ef44..2680e169 100644
--- a/Resources/views/Form/fields.html.twig
+++ b/Resources/views/Form/fields.html.twig
@@ -184,6 +184,17 @@
                 </div>
             </div>
 
+            <div class="control-group control-group-choice">
+                <div class="control-label wrap">
+                    {{ UI.tooltip('oro.akeneo.integration.settings.akeneo_configurable_product_filter.tooltip'|trans, {}, 'right') }}
+                    {{ form_label(form.configurableProductFilter) }}
+                </div>
+                <div class="controls {% if (form_errors(form.configurableProductFilter)|length > 0) %} validation-error {% endif %}">
+                    {{ form_widget(form.configurableProductFilter) }}
+                    {{ form_errors(form.configurableProductFilter) }}
+                </div>
+            </div>
+
             <div class="control-group control-group-choice">
                 <div class="control-label wrap">
                     {{ UI.tooltip('oro.akeneo.integration.settings.akeneo_attribute_list.tooltip'|trans, {}, 'right') }}
diff --git a/Validator/UniqueProductVariantLinksValidator.php b/Validator/UniqueProductVariantLinksValidator.php
index e6e8b001..fa0ded4b 100644
--- a/Validator/UniqueProductVariantLinksValidator.php
+++ b/Validator/UniqueProductVariantLinksValidator.php
@@ -2,8 +2,8 @@
 
 namespace Oro\Bundle\AkeneoBundle\Validator;
 
+use Doctrine\ORM\PersistentCollection;
 use Oro\Bundle\EntityBundle\ORM\DoctrineHelper;
-use Oro\Bundle\ProductBundle\Entity\Product;
 use Oro\Bundle\ProductBundle\Validator\Constraints\ConfigurableProductAccessorTrait;
 use Symfony\Component\Validator\Constraint;
 use Symfony\Component\Validator\ConstraintValidator;
@@ -40,17 +40,14 @@ public function validate($value, Constraint $constraint)
             return;
         }
         if (count($product->getVariantFields()) === 0) {
-            return null;
+            return;
         }
 
-        $uow = $this->doctrineHelper->getEntityManagerForClass(Product::class)->getUnitOfWork();
-        $collections = array_merge($uow->getScheduledCollectionUpdates(), $uow->getScheduledCollectionDeletions());
-        if (
-            !in_array($value->getVariantLinks(), $collections)
-            && !in_array($value->getParentVariantLinks(), $collections)
-            && empty($uow->getEntityChangeSet($value)['variantFields'])
-        ) {
-            return;
+        $variantLinks = $value->getVariantLinks();
+        if ($variantLinks instanceof PersistentCollection) {
+            if ($variantLinks->isInitialized() && !$variantLinks->isDirty()) {
+                return;
+            }
         }
 
         $this->validator->validate($value, $constraint);

From 3e0037f8f7d368085e4577cdcad5957d15643f8f Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Tue, 7 Dec 2021 15:38:24 +0200
Subject: [PATCH 12/14] =?UTF-8?q?The=20identifier=20id=20is=20missing=20fo?=
 =?UTF-8?q?r=20a=20query=20of=20Oro\Bundle\IntegrationBundle\Entity\Channe?=
 =?UTF-8?q?l=E2=80=9D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 ImportExport/EventListener/OwnerStrategyEventListener.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/ImportExport/EventListener/OwnerStrategyEventListener.php b/ImportExport/EventListener/OwnerStrategyEventListener.php
index 5c3dc74c..42719a35 100644
--- a/ImportExport/EventListener/OwnerStrategyEventListener.php
+++ b/ImportExport/EventListener/OwnerStrategyEventListener.php
@@ -37,7 +37,7 @@ public function onProcessBefore(StrategyEvent $event)
 
     protected function getChannel(ContextInterface $context)
     {
-        if (!$this->channel) {
+        if (!$this->channel && $context->getOption('channel')) {
             $this->channel = $this->doctrineHelper->getEntityReference(
                 Channel::class,
                 $context->getOption('channel')

From a32d011f843537ff2d5db5a8ae1b862de796fe5f Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <szhuravel@oroinc.com>
Date: Thu, 26 May 2022 19:10:32 +0300
Subject: [PATCH 13/14] Add cleanup to cron

---
 Command/CleanupCommand.php | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Command/CleanupCommand.php b/Command/CleanupCommand.php
index d4e2e794..071951a0 100644
--- a/Command/CleanupCommand.php
+++ b/Command/CleanupCommand.php
@@ -17,7 +17,7 @@
 class CleanupCommand extends Command implements CronCommandInterface
 {
     /** @var string */
-    protected static $defaultName = 'oro:akeneo:cleanup';
+    protected static $defaultName = 'oro:cron:akeneo:cleanup';
 
     /** @var DoctrineHelper */
     private $doctrineHelper;

From 0a87a9c9c6b998f51c38bbfd6ec0a56034c6b7e5 Mon Sep 17 00:00:00 2001
From: Serhii Zhuravel <1804871+dxops@users.noreply.github.com>
Date: Fri, 3 Jun 2022 23:25:19 +0300
Subject: [PATCH 14/14] AKM-35: Issues in configurable import (#71)

---
 ImportExport/Writer/ConfigurableAsyncWriter.php | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/ImportExport/Writer/ConfigurableAsyncWriter.php b/ImportExport/Writer/ConfigurableAsyncWriter.php
index c1e55588..1a308eec 100644
--- a/ImportExport/Writer/ConfigurableAsyncWriter.php
+++ b/ImportExport/Writer/ConfigurableAsyncWriter.php
@@ -74,11 +74,11 @@ public function write(array $items)
                     }
                 }
 
-                return;
+                continue;
             }
 
             if (empty($item['parent'])) {
-                return;
+                continue;
             }
 
             $parent = $item['parent'];