Introduce a bulk format that uses a prefix length

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/RestNoopBulkAction.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.rest.action.RestBuilderListener;
+import org.elasticsearch.rest.action.document.RestBulkAction;
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
@@ -75,6 +76,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             defaultListExecutedPipelines,
             true,
             request.getXContentType(),
+            RestBulkAction.BulkFormat.MARKER_SUFFIX,
             request.getRestApiVersion()
         );
 

docs/changelog/135506.yaml

@@ -0,0 +1,6 @@
+pr: 135506
+summary: "[Draft] Introduce a bulk format that uses a prefix length"
+area: Store
+type: enhancement
+issues:
+ - 94319

qa/smoke-test-http/src/internalClusterTest/java/org/elasticsearch/http/BulkRestIT.java → qa/smoke-test-http/src/internalClusterTest/java/org/elasticsearch/http/BulkRestMarkerSuffixIT.java

@@ -13,6 +13,7 @@
 import org.apache.http.entity.ContentType;
 import org.elasticsearch.action.bulk.IncrementalBulkService;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.settings.Settings;
@@ -31,7 +32,9 @@
 import static org.hamcrest.Matchers.equalTo;
 
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.SUITE, supportsDedicatedMasters = false, numDataNodes = 2, numClientNodes = 0)
-public class BulkRestIT extends HttpSmokeTestCase {
+public class BulkRestMarkerSuffixIT extends HttpSmokeTestCase {
+
+    private final RequestOptions options = RequestOptions.DEFAULT.toBuilder().addHeader("X-Bulk-Format", "marker-suffix").build();
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
@@ -53,6 +56,9 @@ public void testBulkUriMatchingDoesNotMatchBulkCapabilitiesApi() throws IOExcept
 
     public void testBulkMissingBody() throws IOException {
         Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
+        if (randomBoolean()) {
+            request.setOptions(options);
+        }
         request.setJsonEntity("");
         ResponseException responseException = expectThrows(ResponseException.class, () -> getRestClient().performRequest(request));
         assertEquals(400, responseException.getResponse().getStatusLine().getStatusCode());
@@ -61,6 +67,9 @@ public void testBulkMissingBody() throws IOException {
 
     public void testBulkInvalidIndexNameString() throws IOException {
         Request request = new Request("POST", "/_bulk");
+        if (randomBoolean()) {
+            request.setOptions(options);
+        }
 
         byte[] bytes1 = "{\"create\":{\"_index\":\"".getBytes(StandardCharsets.UTF_8);
         byte[] bytes2 = new byte[] { (byte) 0xfe, (byte) 0xfe, (byte) 0xff, (byte) 0xff };
@@ -80,6 +89,9 @@ public void testBulkInvalidIndexNameString() throws IOException {
 
     public void testBulkRequestBodyImproperlyTerminated() throws IOException {
         Request request = new Request(randomBoolean() ? "POST" : "PUT", "/_bulk");
+        if (randomBoolean()) {
+            request.setOptions(options);
+        }
         // missing final line of the bulk body. cannot process
         request.setJsonEntity(
             "{\"index\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n"
@@ -107,7 +119,9 @@ public void testBulkRequest() throws IOException {
         assertThat(indexCreatedResponse.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
         Request firstBulkRequest = new Request("POST", "/index_name/_bulk");
-
+        if (randomBoolean()) {
+            firstBulkRequest.setOptions(options);
+        }
         String bulkBody = "{\"index\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n"
             + "{\"field\":1}\n"
             + "{\"index\":{\"_index\":\"index_name\",\"_id\":\"2\"}}\n"
@@ -138,6 +152,9 @@ public void testBulkWithIncrementalDisabled() throws IOException {
         assertThat(indexCreatedResponse.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
         Request firstBulkRequest = new Request("POST", "/index_name/_bulk");
+        if (randomBoolean()) {
+            firstBulkRequest.setOptions(options);
+        }
 
         String bulkBody = "{\"index\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n"
             + "{\"field\":1}\n"
@@ -178,6 +195,9 @@ public void testMalformedActionLineBulk() throws IOException {
         assertThat(indexCreatedResponse.getStatusLine().getStatusCode(), equalTo(OK.getStatus()));
 
         Request bulkRequest = new Request("POST", "/index_name/_bulk");
+        if (randomBoolean()) {
+            bulkRequest.setOptions(options);
+        }
 
         final StringBuilder bulk = new StringBuilder();
         bulk.append("{\"index\":{\"_index\":\"index_name\"}}\n");
@@ -191,9 +211,11 @@ public void testMalformedActionLineBulk() throws IOException {
     }
 
     @SuppressWarnings("unchecked")
-    private static void sendLargeBulk() throws IOException {
+    private void sendLargeBulk() throws IOException {
         Request bulkRequest = new Request("POST", "/index_name/_bulk");
-
+        if (randomBoolean()) {
+            bulkRequest.setOptions(options);
+        }
         final StringBuilder bulk = new StringBuilder();
         bulk.append("{\"delete\":{\"_index\":\"index_name\",\"_id\":\"1\"}}\n");
         int updates = 0;