Merge pull request #658 from scireum/ili/OX-11111

Handles large bulk ES requests

src/main/java/sirius/db/es/BulkContext.java

@@ -39,6 +39,7 @@
 public class BulkContext implements Closeable {
 
     private static final int MAX_BATCH_SIZE = 1024;
+    private static final int MAX_REQUEST_SIZE = 100_000_000;
     private static final int RECOMMENDED_BATCH_SIZE = 256;
 
     private static final String KEY_INDEX = "_index";
@@ -128,17 +129,45 @@ private void update(ElasticEntity entity, boolean force) {
             return;
         }
 
-        commands.add(Json.createObject().set(COMMAND_INDEX, meta));
-        commands.add(data);
+        ObjectNode metaCommand = Json.createObject().set(COMMAND_INDEX, meta);
+        commitIfNeededAndAdd(metaCommand, data);
         autocommit();
     }
 
+    /**
+     * Commits all commands if the batch size exceeds {@link #MAX_BATCH_SIZE}.
+     */
     private void autocommit() {
         if (commands.size() >= MAX_BATCH_SIZE) {
             commit().throwFailures();
         }
     }
 
+    /**
+     * Commits queued commands if needed and adds the given nodes to the queue.
+     * <p>
+     * If the new nodes would exceed {@link #MAX_REQUEST_SIZE}, we commit them before adding new commands to the list. This is important
+     * for bulk requests since Elasticsearch establishes a default limit of 100MB per request.
+     *
+     * @param nodes the nodes which size should be checked
+     */
+    private void commitIfNeededAndAdd(ObjectNode... nodes) {
+        int currentSize = commands.stream().map(this::calculateNodeLength).reduce(0, Integer::sum);
+        for (ObjectNode node : nodes) {
+            currentSize += calculateNodeLength(node);
+        }
+        if (currentSize >= MAX_REQUEST_SIZE) {
+            // The given commands will exceed the maximum request size, so we commit the current queued commands
+            // before adding new ones
+            commit().throwFailures();
+        }
+        commands.addAll(List.of(nodes));
+    }
+
+    private int calculateNodeLength(ObjectNode node) {
+        return node.toString().length();
+    }
+
     private ObjectNode builtMetadata(ElasticEntity entity, boolean force, EntityDescriptor ed) {
         ObjectNode meta = Json.createObject();
 
@@ -167,7 +196,9 @@ private void delete(ElasticEntity entity, boolean force) {
         entityDescriptor.beforeDelete(entity);
 
         ObjectNode meta = builtMetadata(entity, force, entityDescriptor);
-        commands.add(Json.createObject().set(COMMAND_DELETE, meta));
+
+        ObjectNode metaCommand = Json.createObject().set(COMMAND_DELETE, meta);
+        commitIfNeededAndAdd(metaCommand);
         autocommit();
     }
 

src/main/java/sirius/db/es/Elastic.java

@@ -12,7 +12,6 @@
 import org.apache.http.HttpHost;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RestClient;
-import org.elasticsearch.client.RestClientBuilder;
 import sirius.db.KeyGenerator;
 import sirius.db.es.constraints.ElasticConstraint;
 import sirius.db.es.constraints.ElasticFilterFactory;
@@ -169,19 +168,14 @@ private synchronized void initializeClient() {
         if (client == null) {
             Elastic.LOG.INFO("Initializing Elasticsearch client against: %s", hosts);
 
-            // Fixes an Elastic bug that results in TimeoutExceptions
-            // Remove this, once ES is updated to at least 6.3.1
-            RestClientBuilder.RequestConfigCallback configCallback =
-                    requestConfigBuilder -> requestConfigBuilder.setConnectionRequestTimeout(0);
-
             HttpHost[] httpHosts = Arrays.stream(this.hosts.split(","))
                                          .map(String::trim)
                                          .map(host -> Strings.splitAtLast(host, ":"))
                                          .map(this::parsePort)
                                          .map(this::mapPort)
                                          .map(this::makeHttpHost)
                                          .toArray(size -> new HttpHost[size]);
-            client = new LowLevelClient(RestClient.builder(httpHosts).setRequestConfigCallback(configCallback).build());
+            client = new LowLevelClient(RestClient.builder(httpHosts).build());
 
             // If we're using a docker container (most probably for testing), we give ES some time
             // to fully boot up. Otherwise, strange connection issues might arise.

src/main/java/sirius/db/es/RequestBuilder.java

@@ -246,7 +246,11 @@ private HandledException handleAsyncFailure(Exception exception, String uri) {
 
     protected ObjectNode extractErrorJSON(ResponseException e) {
         try {
-            ObjectNode response = Json.parseObject(EntityUtils.toString(e.getResponse().getEntity()));
+            HttpEntity httpEntity = e.getResponse().getEntity();
+            if (e.getResponse().getEntity().getContentLength() == 0) {
+                return null;
+            }
+            ObjectNode response = Json.parseObject(EntityUtils.toString(httpEntity));
             return Json.getObject(response, PARAM_ERROR);
         } catch (IOException ex) {
             Exceptions.handle(Elastic.LOG, ex);