Use errors output field to determine _bulk API failure

Elastic search _bulk API's response HTTP status code 200 does not imply
that the request was successful. Instead, depend on 'errors' boolean
from response body [1]. Also add tests to illustrate this behaviour.

[1] https://www.elastic.co/guide/en/elasticsearch/reference/7.8/docs-bulk.html#bulk-api-response-body

Change-Id: I6b76398d64efc15803b9a1b30e4d88ae4b1bdb2a

src/main/java/com/google/gerrit/elasticsearch/AbstractElasticIndex.java

diff --git a/src/main/java/com/google/gerrit/elasticsearch/AbstractElasticIndex.java b/src/main/java/com/google/gerrit/elasticsearch/AbstractElasticIndex.java
index c6318fa..d87f993 100644
--- a/src/main/java/com/google/gerrit/elasticsearch/AbstractElasticIndex.java
+++ b/src/main/java/com/google/gerrit/elasticsearch/AbstractElasticIndex.java
@@ -19,6 +19,7 @@
 import static com.google.gson.FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -75,6 +76,7 @@
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.ContentType;
 import org.apache.http.nio.entity.NStringEntity;
+import org.apache.http.util.EntityUtils;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 
@@ -271,6 +273,20 @@
     return new FieldBundle(rawFields);
   }
 
+  protected boolean hasErrors(Response response) {
+    try {
+      String contentType = response.getEntity().getContentType().getValue();
+      Preconditions.checkState(
+          contentType.equals(ContentType.APPLICATION_JSON.toString()),
+          String.format("Expected %s, but was: %s", ContentType.APPLICATION_JSON, contentType));
+      String responseStr = EntityUtils.toString(response.getEntity());
+      JsonObject responseJson = (JsonObject) new JsonParser().parse(responseStr);
+      return responseJson.get("errors").getAsBoolean();
+    } catch (IOException e) {
+      throw new StorageException(e);
+    }
+  }
+
   protected String toAction(String type, String id, String action) {
     JsonObject properties = new JsonObject();
     properties.addProperty("_id", id);

src/main/java/com/google/gerrit/elasticsearch/ElasticAccountIndex.java

diff --git a/src/main/java/com/google/gerrit/elasticsearch/ElasticAccountIndex.java b/src/main/java/com/google/gerrit/elasticsearch/ElasticAccountIndex.java
index c350fcc..1da80ae 100644
--- a/src/main/java/com/google/gerrit/elasticsearch/ElasticAccountIndex.java
+++ b/src/main/java/com/google/gerrit/elasticsearch/ElasticAccountIndex.java
@@ -82,7 +82,7 @@
     String uri = getURI(BULK);
     Response response = postRequestWithRefreshParam(uri, bulk);
     int statusCode = response.getStatusLine().getStatusCode();
-    if (statusCode != HttpStatus.SC_OK) {
+    if (hasErrors(response) || statusCode != HttpStatus.SC_OK) {
       throw new StorageException(
           String.format(
               "Failed to replace account %s in index %s: %s",

src/main/java/com/google/gerrit/elasticsearch/ElasticChangeIndex.java

diff --git a/src/main/java/com/google/gerrit/elasticsearch/ElasticChangeIndex.java b/src/main/java/com/google/gerrit/elasticsearch/ElasticChangeIndex.java
index c0ac239..a4674bf 100644
--- a/src/main/java/com/google/gerrit/elasticsearch/ElasticChangeIndex.java
+++ b/src/main/java/com/google/gerrit/elasticsearch/ElasticChangeIndex.java
@@ -102,7 +102,7 @@
     String uri = getURI(BULK);
     Response response = postRequestWithRefreshParam(uri, bulk);
     int statusCode = response.getStatusLine().getStatusCode();
-    if (statusCode != HttpStatus.SC_OK) {
+    if (hasErrors(response) || statusCode != HttpStatus.SC_OK) {
       throw new StorageException(
           String.format(
               "Failed to replace change %s in index %s: %s", cd.getId(), indexName, statusCode));

src/main/java/com/google/gerrit/elasticsearch/ElasticGroupIndex.java

diff --git a/src/main/java/com/google/gerrit/elasticsearch/ElasticGroupIndex.java b/src/main/java/com/google/gerrit/elasticsearch/ElasticGroupIndex.java
index f27c37f..626203a 100644
--- a/src/main/java/com/google/gerrit/elasticsearch/ElasticGroupIndex.java
+++ b/src/main/java/com/google/gerrit/elasticsearch/ElasticGroupIndex.java
@@ -82,7 +82,7 @@
     String uri = getURI(BULK);
     Response response = postRequestWithRefreshParam(uri, bulk);
     int statusCode = response.getStatusLine().getStatusCode();
-    if (statusCode != HttpStatus.SC_OK) {
+    if (hasErrors(response) || statusCode != HttpStatus.SC_OK) {
       throw new StorageException(
           String.format(
               "Failed to replace group %s in index %s: %s",

src/main/java/com/google/gerrit/elasticsearch/ElasticProjectIndex.java

diff --git a/src/main/java/com/google/gerrit/elasticsearch/ElasticProjectIndex.java b/src/main/java/com/google/gerrit/elasticsearch/ElasticProjectIndex.java
index 0ccc954..99b202d 100644
--- a/src/main/java/com/google/gerrit/elasticsearch/ElasticProjectIndex.java
+++ b/src/main/java/com/google/gerrit/elasticsearch/ElasticProjectIndex.java
@@ -84,7 +84,7 @@
     String uri = getURI(BULK);
     Response response = postRequestWithRefreshParam(uri, bulk);
     int statusCode = response.getStatusLine().getStatusCode();
-    if (statusCode != HttpStatus.SC_OK) {
+    if (hasErrors(response) || statusCode != HttpStatus.SC_OK) {
       throw new StorageException(
           String.format(
               "Failed to replace project %s in index %s: %s",

src/test/java/com/google/gerrit/elasticsearch/ElasticTestUtils.java

diff --git a/src/test/java/com/google/gerrit/elasticsearch/ElasticTestUtils.java b/src/test/java/com/google/gerrit/elasticsearch/ElasticTestUtils.java
index c3ca595..f5ba9db 100644
--- a/src/test/java/com/google/gerrit/elasticsearch/ElasticTestUtils.java
+++ b/src/test/java/com/google/gerrit/elasticsearch/ElasticTestUtils.java
@@ -14,6 +14,8 @@
 
 package com.google.gerrit.elasticsearch;
 
+import static java.util.concurrent.TimeUnit.MINUTES;
+
 import com.google.gerrit.index.IndexDefinition;
 import com.google.gerrit.server.LibModuleType;
 import com.google.gerrit.testing.GerritTestName;
@@ -25,6 +27,9 @@
 import com.google.inject.TypeLiteral;
 import java.util.Collection;
 import java.util.UUID;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
 import org.eclipse.jgit.lib.Config;
 
 public final class ElasticTestUtils {
@@ -81,6 +86,22 @@
     return Guice.createInjector(new InMemoryModule(elasticsearchConfig));
   }
 
+  public static void closeIndex(
+      CloseableHttpAsyncClient client, ElasticContainer container, GerritTestName testName)
+      throws Exception {
+    client
+        .execute(
+            new HttpPost(
+                String.format(
+                    "http://%s:%d/%s*/_close",
+                    container.getHttpHost().getHostName(),
+                    container.getHttpHost().getPort(),
+                    testName.getSanitizedMethodName())),
+            HttpClientContext.create(),
+            null)
+        .get(5, MINUTES);
+  }
+
   private ElasticTestUtils() {
     // hide default constructor
   }

src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryAccountsTest.java

diff --git a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryAccountsTest.java b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryAccountsTest.java
index 4ee5a16..30d07c7 100644
--- a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryAccountsTest.java
+++ b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryAccountsTest.java
@@ -14,12 +14,19 @@
 
 package com.google.gerrit.elasticsearch;
 
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+
+import com.google.gerrit.exceptions.StorageException;
 import com.google.gerrit.server.query.account.AbstractQueryAccountsTest;
 import com.google.gerrit.testing.ConfigSuite;
 import com.google.inject.Injector;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
 import org.eclipse.jgit.lib.Config;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 public class ElasticV7QueryAccountsTest extends AbstractQueryAccountsTest {
   @ConfigSuite.Default
@@ -28,12 +35,15 @@
   }
 
   private static ElasticContainer container;
+  private static CloseableHttpAsyncClient client;
 
   @BeforeClass
   public static void startIndexService() {
     if (container == null) {
       // Only start Elasticsearch once
       container = ElasticContainer.createAndStart(ElasticVersion.V7_8);
+      client = HttpAsyncClients.createDefault();
+      client.start();
     }
   }
 
@@ -54,4 +64,14 @@
   protected Injector createInjector() {
     return ElasticTestUtils.createInjector(config, testName, container);
   }
+
+  @Test
+  public void testErrorResponseFromAccountIndex() throws Exception {
+    gApi.accounts().self().index();
+
+    ElasticTestUtils.closeIndex(client, container, testName);
+    StorageException thrown =
+        assertThrows(StorageException.class, () -> gApi.accounts().self().index());
+    assertThat(thrown).hasMessageThat().contains("Failed to replace account");
+  }
 }

src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryChangesTest.java

diff --git a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryChangesTest.java b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryChangesTest.java
index 6a0ce76..e7f1d1c 100644
--- a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryChangesTest.java
+++ b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryChangesTest.java
@@ -14,21 +14,25 @@
 
 package com.google.gerrit.elasticsearch;
 
-import static java.util.concurrent.TimeUnit.MINUTES;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
 
+import com.google.gerrit.entities.Change;
+import com.google.gerrit.exceptions.StorageException;
 import com.google.gerrit.server.query.change.AbstractQueryChangesTest;
 import com.google.gerrit.testing.ConfigSuite;
 import com.google.gerrit.testing.GerritTestName;
+import com.google.gerrit.testing.InMemoryRepositoryManager;
 import com.google.inject.Injector;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.client.protocol.HttpClientContext;
 import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
 import org.apache.http.impl.nio.client.HttpAsyncClients;
+import org.eclipse.jgit.junit.TestRepository;
 import org.eclipse.jgit.lib.Config;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Rule;
+import org.junit.Test;
 
 public class ElasticV7QueryChangesTest extends AbstractQueryChangesTest {
   @ConfigSuite.Default
@@ -62,17 +66,7 @@
   public void closeIndex() throws Exception {
     // Close the index after each test to prevent exceeding Elasticsearch's
     // shard limit (see Issue 10120).
-    client
-        .execute(
-            new HttpPost(
-                String.format(
-                    "http://%s:%d/%s*/_close",
-                    container.getHttpHost().getHostName(),
-                    container.getHttpHost().getPort(),
-                    testName.getSanitizedMethodName())),
-            HttpClientContext.create(),
-            null)
-        .get(5, MINUTES);
+    ElasticTestUtils.closeIndex(client, container, testName);
   }
 
   @Override
@@ -85,4 +79,16 @@
   protected Injector createInjector() {
     return ElasticTestUtils.createInjector(config, testName, container);
   }
+
+  @Test
+  public void testErrorResponseFromChangeIndex() throws Exception {
+    TestRepository<InMemoryRepositoryManager.Repo> repo = createProject("repo");
+    Change c = insert(repo, newChangeWithStatus(repo, Change.Status.NEW));
+    gApi.changes().id(c.getChangeId()).index();
+
+    ElasticTestUtils.closeIndex(client, container, testName);
+    StorageException thrown =
+        assertThrows(StorageException.class, () -> gApi.changes().id(c.getChangeId()).index());
+    assertThat(thrown).hasMessageThat().contains("Failed to reindex change");
+  }
 }

src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryGroupsTest.java

diff --git a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryGroupsTest.java b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryGroupsTest.java
index 649c0bc..17fc0d2 100644
--- a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryGroupsTest.java
+++ b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryGroupsTest.java
@@ -14,12 +14,20 @@
 
 package com.google.gerrit.elasticsearch;
 
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+
+import com.google.gerrit.exceptions.StorageException;
+import com.google.gerrit.extensions.api.groups.GroupApi;
 import com.google.gerrit.server.query.group.AbstractQueryGroupsTest;
 import com.google.gerrit.testing.ConfigSuite;
 import com.google.inject.Injector;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
 import org.eclipse.jgit.lib.Config;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 public class ElasticV7QueryGroupsTest extends AbstractQueryGroupsTest {
   @ConfigSuite.Default
@@ -28,12 +36,15 @@
   }
 
   private static ElasticContainer container;
+  private static CloseableHttpAsyncClient client;
 
   @BeforeClass
   public static void startIndexService() {
     if (container == null) {
       // Only start Elasticsearch once
       container = ElasticContainer.createAndStart(ElasticVersion.V7_8);
+      client = HttpAsyncClients.createDefault();
+      client.start();
     }
   }
 
@@ -54,4 +65,14 @@
   protected Injector createInjector() {
     return ElasticTestUtils.createInjector(config, testName, container);
   }
+
+  @Test
+  public void testErrorResponseFromGroupIndex() throws Exception {
+    GroupApi group = gApi.groups().create("test");
+    group.index();
+
+    ElasticTestUtils.closeIndex(client, container, testName);
+    StorageException thrown = assertThrows(StorageException.class, () -> group.index());
+    assertThat(thrown).hasMessageThat().contains("Failed to replace group");
+  }
 }

src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryProjectsTest.java

diff --git a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryProjectsTest.java b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryProjectsTest.java
index d3b3d44..f5524d7 100644
--- a/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryProjectsTest.java
+++ b/src/test/java/com/google/gerrit/elasticsearch/ElasticV7QueryProjectsTest.java
@@ -14,12 +14,20 @@
 
 package com.google.gerrit.elasticsearch;
 
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+
+import com.google.gerrit.exceptions.StorageException;
+import com.google.gerrit.extensions.api.projects.ProjectApi;
 import com.google.gerrit.server.query.project.AbstractQueryProjectsTest;
 import com.google.gerrit.testing.ConfigSuite;
 import com.google.inject.Injector;
+import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
+import org.apache.http.impl.nio.client.HttpAsyncClients;
 import org.eclipse.jgit.lib.Config;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 public class ElasticV7QueryProjectsTest extends AbstractQueryProjectsTest {
   @ConfigSuite.Default
@@ -28,12 +36,15 @@
   }
 
   private static ElasticContainer container;
+  private static CloseableHttpAsyncClient client;
 
   @BeforeClass
   public static void startIndexService() {
     if (container == null) {
       // Only start Elasticsearch once
       container = ElasticContainer.createAndStart(ElasticVersion.V7_8);
+      client = HttpAsyncClients.createDefault();
+      client.start();
     }
   }
 
@@ -54,4 +65,14 @@
   protected Injector createInjector() {
     return ElasticTestUtils.createInjector(config, testName, container);
   }
+
+  @Test
+  public void testErrorResponseFromProjectIndex() throws Exception {
+    ProjectApi project = gApi.projects().create("test");
+    project.index(false);
+
+    ElasticTestUtils.closeIndex(client, container, testName);
+    StorageException thrown = assertThrows(StorageException.class, () -> project.index(false));
+    assertThat(thrown).hasMessageThat().contains("Failed to replace project");
+  }
 }