MAGETWO-93966: Run Catalog Search reindex in multithread mode with Elasticsearch 5

Author: slopukhov

app/code/Magento/Elasticsearch/Elasticsearch5/Model/Client/Elasticsearch.php

@@ -3,7 +3,6 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
-
 namespace Magento\Elasticsearch\Elasticsearch5\Model\Client;
 
 use Magento\Framework\Exception\LocalizedException;
@@ -17,7 +16,7 @@ class Elasticsearch implements ClientInterface
     /**
      * Elasticsearch Client instance
      *
-     * @var \Elasticsearch\Client
+     * @var \Elasticsearch\Client[]
      */
     protected $client;
 
@@ -48,7 +47,7 @@ public function __construct(
         $elasticsearchClient = null
     ) {
         if (empty($options['hostname']) || ((!empty($options['enableAuth']) &&
-            ($options['enableAuth'] == 1)) && (empty($options['username']) || empty($options['password'])))) {
+                    ($options['enableAuth'] == 1)) && (empty($options['username']) || empty($options['password'])))) {
             throw new LocalizedException(
                 __('The search failed because of a search engine misconfiguration.')
             );
@@ -58,10 +57,25 @@ public function __construct(
             $config = $this->buildConfig($options);
             $elasticsearchClient = \Elasticsearch\ClientBuilder::fromConfig($config, true);
         }
-        $this->client = $elasticsearchClient;
+        $this->client[getmypid()] = $elasticsearchClient;
         $this->clientOptions = $options;
     }
 
+    /**
+     * Get Elasticsearch Client
+     *
+     * @return \Elasticsearch\Client
+     */
+    private function getClient()
+    {
+        $pid = getmypid();
+        if (!isset($this->client[$pid])) {
+            $config = $this->buildConfig($this->clientOptions);
+            $this->client[$pid] = \Elasticsearch\ClientBuilder::fromConfig($config, true);
+        }
+        return $this->client[$pid];
+    }
+
     /**
      * Ping the Elasticsearch client
      *
@@ -70,7 +84,7 @@ public function __construct(
     public function ping()
     {
         if ($this->pingResult === null) {
-            $this->pingResult = $this->client->ping(['client' => ['timeout' => $this->clientOptions['timeout']]]);
+            $this->pingResult = $this->getClient()->ping(['client' => ['timeout' => $this->clientOptions['timeout']]]);
         }
 
         return $this->pingResult;
@@ -116,7 +130,7 @@ private function buildConfig($options = [])
      */
     public function bulkQuery($query)
     {
-        $this->client->bulk($query);
+        $this->getClient()->bulk($query);
     }
 
     /**
@@ -128,7 +142,7 @@ public function bulkQuery($query)
      */
     public function createIndex($index, $settings)
     {
-        $this->client->indices()->create([
+        $this->getClient()->indices()->create([
             'index' => $index,
             'body' => $settings,
         ]);
@@ -142,7 +156,7 @@ public function createIndex($index, $settings)
      */
     public function deleteIndex($index)
     {
-        $this->client->indices()->delete(['index' => $index]);
+        $this->getClient()->indices()->delete(['index' => $index]);
     }
 
     /**
@@ -153,7 +167,7 @@ public function deleteIndex($index)
      */
     public function isEmptyIndex($index)
     {
-        $stats = $this->client->indices()->stats(['index' => $index, 'metric' => 'docs']);
+        $stats = $this->getClient()->indices()->stats(['index' => $index, 'metric' => 'docs']);
         if ($stats['indices'][$index]['primaries']['docs']['count'] == 0) {
             return true;
         }
@@ -178,7 +192,7 @@ public function updateAlias($alias, $newIndex, $oldIndex = '')
             $params['body']['actions'][] = ['add' => ['alias' => $alias, 'index' => $newIndex]];
         }
 
-        $this->client->indices()->updateAliases($params);
+        $this->getClient()->indices()->updateAliases($params);
     }
 
     /**
@@ -189,7 +203,7 @@ public function updateAlias($alias, $newIndex, $oldIndex = '')
      */
     public function indexExists($index)
     {
-         return $this->client->indices()->exists(['index' => $index]);
+        return $this->getClient()->indices()->exists(['index' => $index]);
     }
 
     /**
@@ -204,7 +218,7 @@ public function existsAlias($alias, $index = '')
         if ($index) {
             $params['index'] = $index;
         }
-        return $this->client->indices()->existsAlias($params);
+        return $this->getClient()->indices()->existsAlias($params);
     }
 
     /**
@@ -214,7 +228,7 @@ public function existsAlias($alias, $index = '')
      */
     public function getAlias($alias)
     {
-        return $this->client->indices()->getAlias(['name' => $alias]);
+        return $this->getClient()->indices()->getAlias(['name' => $alias]);
     }
 
     /**
@@ -274,7 +288,7 @@ public function addFieldsMapping(array $fields, $index, $entityType)
             $params['body'][$entityType]['properties'][$field] = $this->prepareFieldInfo($fieldInfo);
         }
 
-        $this->client->indices()->putMapping($params);
+        $this->getClient()->indices()->putMapping($params);
     }
 
     /**
@@ -311,7 +325,7 @@ private function prepareFieldInfo($fieldInfo)
      */
     public function deleteMapping($index, $entityType)
     {
-        $this->client->indices()->deleteMapping([
+        $this->getClient()->indices()->deleteMapping([
             'index' => $index,
             'type' => $entityType,
         ]);
@@ -327,7 +341,7 @@ public function query($query)
     {
         $query = $this->prepareSearchQuery($query);
 
-        return $this->client->search($query);
+        return $this->getClient()->search($query);
     }
 
     /**
@@ -358,7 +372,7 @@ private function prepareSearchQuery($query)
      */
     public function suggest($query)
     {
-        return $this->client->suggest($query);
+        return $this->getClient()->suggest($query);
     }
 
     /**
@@ -369,7 +383,7 @@ public function suggest($query)
     private function getServerVersion()
     {
         if ($this->serverVersion === null) {
-            $info = $this->client->info();
+            $info = $this->getClient()->info();
             $this->serverVersion = $info['version']['number'];
         }