Increase the maximum JSON string length in the OpenSearch source and sink to 64 MB

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java

@@ -9,7 +9,6 @@
 
 package org.opensearch.dataprepper.plugins.sink.opensearch;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
@@ -179,7 +178,7 @@ public BulkIngester(final OpenSearchSinkConfiguration openSearchSinkConfig,
         this.versionExpression = openSearchSinkConfig.getIndexConfiguration().getVersionExpression();
         this.scriptManager = new ScriptManager(openSearchSinkConfig.getIndexConfiguration().getScriptConfiguration(),
                 expressionEvaluator);
-        this.bulkOperationFactory = new BulkOperationFactory(versionType, scriptManager, new ObjectMapper(),
+        this.bulkOperationFactory = new BulkOperationFactory(versionType, scriptManager,
                 isUsingDocumentFilters());
         this.failedBulkOperationConverter = new FailedBulkOperationConverter(pipeline, PLUGIN_NAME);
 

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactory.java

@@ -19,6 +19,7 @@
 import org.opensearch.client.opensearch.core.bulk.DeleteOperation;
 import org.opensearch.client.opensearch.core.bulk.IndexOperation;
 import org.opensearch.client.opensearch.core.bulk.UpdateOperation;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 import org.opensearch.dataprepper.model.opensearch.OpenSearchBulkActions;
 import org.opensearch.dataprepper.plugins.sink.opensearch.bulk.SerializedJson;
 
@@ -36,11 +37,10 @@ public class BulkOperationFactory {
 
     public BulkOperationFactory(final VersionType versionType,
                                 final ScriptManager scriptManager,
-                                final ObjectMapper objectMapper,
                                 final boolean usingDocumentFilters) {
         this.versionType = versionType;
         this.scriptManager = scriptManager;
-        this.objectMapper = objectMapper;
+        this.objectMapper = new ObjectMapper(JsonFactories.defaultJsonFactory());
         this.usingDocumentFilters = usingDocumentFilters;
     }
 

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -9,7 +9,6 @@
 
 package org.opensearch.dataprepper.plugins.sink.opensearch;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.opensearch.client.RestHighLevelClient;
 import org.opensearch.client.opensearch.OpenSearchClient;
 import org.opensearch.client.transport.TransportOptions;
@@ -187,7 +186,7 @@ private void doInitializeInternal() throws IOException {
       final PullEngine pullEngine = pluginFactory.loadPlugin(PullEngine.class, enginePluginSetting);
       final IndexShardProvider indexShardProvider = new IndexShardProvider(restHighLevelClient);
       final IndexRouter indexRouter = new IndexRouter(indexShardProvider);
-      final PullIngestionEnvelopeBuilder envelopeBuilder = new PullIngestionEnvelopeBuilder(new ObjectMapper());
+      final PullIngestionEnvelopeBuilder envelopeBuilder = new PullIngestionEnvelopeBuilder();
       final DocumentIdResolver documentIdResolver = new DocumentIdResolver(
               openSearchSinkConfig.getIndexConfiguration().getDocumentIdField(),
               openSearchSinkConfig.getIndexConfiguration().getDocumentId(),

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapper.java

@@ -12,6 +12,7 @@
 import org.opensearch.client.json.jackson.JacksonJsonProvider;
 import org.opensearch.client.json.jackson.JacksonJsonpGenerator;
 import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 
 import java.io.IOException;
 import java.io.OutputStream;
@@ -29,7 +30,7 @@ public class PreSerializedJsonpMapper implements JsonpMapper {
 
     public PreSerializedJsonpMapper() {
         innerMapper = new JacksonJsonpMapper();
-        jsonProvider = new PreSerializedJsonProvider();
+        jsonProvider = new PreSerializedJsonProvider(JsonFactories.defaultJsonFactory());
     }
 
     @Override
@@ -71,7 +72,15 @@ public PreSerializedJsonGenerator(com.fasterxml.jackson.core.JsonGenerator gener
     }
 
 
+    /**
+     * The opensearch-java transports parse responses with parsers created from this provider,
+     * so its JsonFactory carries Data Prepper's maximum JSON string length.
+     */
     private static class PreSerializedJsonProvider extends JacksonJsonProvider {
+        PreSerializedJsonProvider(final com.fasterxml.jackson.core.JsonFactory jsonFactory) {
+            super(jsonFactory);
+        }
+
         @Override
         public JsonGenerator createGenerator(OutputStream out) {
             try {

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverter.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import org.opensearch.client.opensearch.core.bulk.BulkResponseItem;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 import org.opensearch.dataprepper.model.failures.DlqObject;
 import org.opensearch.dataprepper.plugins.sink.opensearch.bulk.SerializedJson;
 import org.opensearch.dataprepper.plugins.sink.opensearch.BulkOperationWrapper;
@@ -15,7 +16,7 @@
 
 public class FailedBulkOperationConverter {
 
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(JsonFactories.defaultJsonFactory());
 
     private final String pluginName;
     private final String pipelineName;

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilder.java

@@ -11,6 +11,7 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -22,8 +23,8 @@ public class PullIngestionEnvelopeBuilder {
 
     private final ObjectMapper objectMapper;
 
-    public PullIngestionEnvelopeBuilder(final ObjectMapper objectMapper) {
-        this.objectMapper = objectMapper;
+    public PullIngestionEnvelopeBuilder() {
+        this.objectMapper = new ObjectMapper(JsonFactories.defaultJsonFactory());
     }
 
     public byte[] build(final String id, final long version, final String source, final String opType) throws JsonProcessingException {

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java

@@ -7,6 +7,8 @@
 
 import co.elastic.clients.elasticsearch.ElasticsearchClient;
 import co.elastic.clients.transport.ElasticsearchTransport;
+import com.fasterxml.jackson.core.StreamReadConstraints;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.http.Header;
 import org.apache.http.HttpHost;
 import org.apache.http.HttpRequestInterceptor;
@@ -29,6 +31,7 @@
 import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions;
 import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
 import org.opensearch.dataprepper.aws.api.AwsRequestSigningApache4Interceptor;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 import org.opensearch.dataprepper.plugins.source.opensearch.AuthConfig;
 import org.opensearch.dataprepper.plugins.certificate.validation.PemObjectValidator;
 import org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSourceConfiguration;
@@ -82,7 +85,7 @@ public ElasticsearchClient provideElasticSearchClient(final OpenSearchSourceConf
     }
 
     private OpenSearchTransport createOpenSearchTransport(final RestClient restClient) {
-        return new RestClientTransport(restClient, new JacksonJsonpMapper());
+        return new RestClientTransport(restClient, createJacksonJsonpMapper());
     }
 
     private OpenSearchTransport createOpenSearchTransportForAws(final OpenSearchSourceConfiguration openSearchSourceConfiguration) {
@@ -102,11 +105,35 @@ private OpenSearchTransport createOpenSearchTransportForAws(final OpenSearchSour
                 openSearchSourceConfiguration.getAwsAuthenticationOptions().getAwsRegion(),
                 AwsSdk2TransportOptions.builder()
                         .setCredentials(awsCredentialsProvider)
-                        .setMapper(new JacksonJsonpMapper())
+                        .setMapper(createJacksonJsonpMapper())
                         .build());
 
     }
 
+    static JacksonJsonpMapper createJacksonJsonpMapper() {
+        final JacksonJsonpMapper jacksonJsonpMapper = new JacksonJsonpMapper();
+        raiseMaxStringLength(jacksonJsonpMapper.objectMapper());
+        return jacksonJsonpMapper;
+    }
+
+    static co.elastic.clients.json.jackson.JacksonJsonpMapper createElasticsearchJacksonJsonpMapper() {
+        final co.elastic.clients.json.jackson.JacksonJsonpMapper jacksonJsonpMapper = new co.elastic.clients.json.jackson.JacksonJsonpMapper();
+        raiseMaxStringLength(jacksonJsonpMapper.objectMapper());
+        return jacksonJsonpMapper;
+    }
+
+    /**
+     * Raises the maximum JSON string length to {@link JsonFactories#MAX_JSON_STRING_LENGTH} so that
+     * documents which can be represented as an Event can also be read from the source cluster.
+     * The JacksonJsonpMapper JSON provider is created from the ObjectMapper's JsonFactory,
+     * so updating the factory's constraints applies to response parsing as well.
+     */
+    private static void raiseMaxStringLength(final ObjectMapper objectMapper) {
+        objectMapper.getFactory().setStreamReadConstraints(StreamReadConstraints.builder()
+                .maxStringLength(JsonFactories.MAX_JSON_STRING_LENGTH)
+                .build());
+    }
+
     public SdkAsyncHttpClient createSdkAsyncHttpClient(final OpenSearchSourceConfiguration openSearchSourceConfiguration) {
         final NettyNioAsyncHttpClient.Builder builder = NettyNioAsyncHttpClient.builder();
 
@@ -139,7 +166,7 @@ private RestClient createOpenSearchRestClient(final OpenSearchSourceConfiguratio
     }
 
     private ElasticsearchTransport createElasticSearchTransport(final org.elasticsearch.client.RestClient restClient) {
-        return new co.elastic.clients.transport.rest_client.RestClientTransport(restClient, new co.elastic.clients.json.jackson.JacksonJsonpMapper());
+        return new co.elastic.clients.transport.rest_client.RestClientTransport(restClient, createElasticsearchJacksonJsonpMapper());
     }
 
     private org.elasticsearch.client.RestClient createElasticSearchRestClient(final OpenSearchSourceConfiguration openSearchSourceConfiguration) {

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactoryTest.java

@@ -31,7 +31,7 @@ public class BulkOperationFactoryTest {
     @BeforeEach
     void setUp() {
         final ScriptManager scriptManager = new ScriptManager(null, null);
-        factory = new BulkOperationFactory(VersionType.External, scriptManager, objectMapper, false);
+        factory = new BulkOperationFactory(VersionType.External, scriptManager, false);
     }
 
     @Test
@@ -126,7 +126,7 @@ void create_action_without_optional_fields() {
     @Test
     void update_with_document_filters_deserializes_from_serialized_json() {
         final ScriptManager scriptManager = new ScriptManager(null, null);
-        final BulkOperationFactory filterFactory = new BulkOperationFactory(null, scriptManager, objectMapper, true);
+        final BulkOperationFactory filterFactory = new BulkOperationFactory(null, scriptManager, true);
         final ObjectNode jsonNode = objectMapper.createObjectNode().put("name", "original");
         final SerializedJson document = SerializedJson.fromStringAndOptionals("{\"name\":\"filtered\"}", "doc-1", null, null);
 
@@ -158,4 +158,18 @@ void index_action_sets_pipeline() {
         assertThat(result.isIndex(), is(true));
         assertThat(result.index().pipeline(), equalTo("my-pipeline"));
     }
+
+    @Test
+    void update_with_document_filters_parses_string_values_above_the_jackson_default_limit() {
+        final int length = 21 * 1024 * 1024;
+        final BulkOperationFactory filterFactory = new BulkOperationFactory(null, new ScriptManager(null, null), true);
+        final SerializedJson document = SerializedJson.fromStringAndOptionals(
+                "{\"largeField\":\"" + "a".repeat(length) + "\"}", "doc-1", null, null);
+
+        final BulkOperation result = filterFactory.create(
+                OpenSearchBulkActions.UPSERT.toString(), document, null, "test-index", objectMapper.createObjectNode());
+
+        assertThat(result.isUpdate(), is(true));
+        assertThat(result.update().id(), equalTo("doc-1"));
+    }
 }

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java

@@ -6,7 +6,6 @@
 package org.opensearch.dataprepper.plugins.sink.opensearch;
 
 import io.micrometer.core.instrument.Measurement;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -1405,7 +1404,7 @@ private static BulkResponseItem bulkFailureWithReason(final String index, final
     private TimeSeriesIndex createTimeSeriesIndex() {
         return new TimeSeriesIndex(
                 new TimeSeriesIndexDetector(mock(OpenSearchClient.class), new IndexCache()),
-                new BulkOperationFactory(null, mock(ScriptManager.class), new ObjectMapper(), false),
+                new BulkOperationFactory(null, mock(ScriptManager.class), false),
                 false);
     }
 

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkScriptTest.java

@@ -39,7 +39,7 @@ public class OpenSearchSinkScriptTest {
 
     private BulkOperationFactory createFactory(final ScriptConfiguration scriptConfig) {
         final ScriptManager scriptManager = new ScriptManager(scriptConfig, null);
-        return new BulkOperationFactory(null, scriptManager, new ObjectMapper(), false);
+        return new BulkOperationFactory(null, scriptManager, false);
     }
 
     private ScriptConfiguration mockScript(final String source, final Map<String, Object> params) {

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapperTest.java

@@ -8,6 +8,8 @@
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.StringReader;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import jakarta.json.stream.JsonGenerator;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -111,4 +113,16 @@ void serialize_on_Map_uses_Jackson_serializer() throws IOException {
 
         assertThat(new String(outputStream.toByteArray()), equalTo(expectedSerializedJson));
     }
-}
+
+    @Test
+    void deserialize_parses_string_values_above_the_jackson_default_limit() {
+        final int length = 21 * 1024 * 1024;
+        final String json = "{\"largeField\":\"" + "a".repeat(length) + "\"}";
+        final PreSerializedJsonpMapper objectUnderTest = createObjectUnderTest();
+
+        try (final jakarta.json.stream.JsonParser parser = objectUnderTest.jsonProvider().createParser(new StringReader(json))) {
+            final ObjectNode result = objectUnderTest.deserialize(parser, ObjectNode.class);
+            assertThat(result.get("largeField").textValue().length(), equalTo(length));
+        }
+    }
+}

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverterTest.java

@@ -179,5 +179,26 @@ private void generateRandomDocument() {
         document = SerializedJson.fromStringAndOptionals(jsonString, null, null, null);
     }
 
+    @Test
+    public void testConvertToDlqObjectWithStringValueAboveTheJacksonDefaultLimit() {
+        final int length = 21 * 1024 * 1024;
+        final SerializedJson largeDocument = SerializedJson.fromStringAndOptionals(
+                "{\"largeField\":\"" + "a".repeat(length) + "\"}", null, null, null);
+        final IndexOperation indexOperation = mock(IndexOperation.class);
+        when(indexOperation.index()).thenReturn(testIndex);
+        when(indexOperation.id()).thenReturn(testId);
+        when(indexOperation.document()).thenReturn(largeDocument);
+        when(bulkOperation.index()).thenReturn(indexOperation);
+
+        final FailedBulkOperation testData = FailedBulkOperation.builder()
+                .withBulkOperation(new BulkOperationWrapper(bulkOperation))
+                .withBulkResponseItem(bulkResponseItem)
+                .withFailure(failure)
+                .build();
 
+        final DlqObject result = converter.convertToDlqObject(testData);
+
+        final Map<String, Object> doc = (Map<String, Object>) ((FailedDlqData) result.getFailedData()).getDocument();
+        assertThat(((String) doc.get("largeField")).length(), is(equalTo(length)));
+    }
 }

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/TimeSeriesIndexTest.java

@@ -65,7 +65,7 @@ class TimeSeriesIndexTest {
 
     @BeforeEach
     void setUp() {
-        bulkOperationFactory = new BulkOperationFactory(null, mock(ScriptManager.class), new ObjectMapper(), false);
+        bulkOperationFactory = new BulkOperationFactory(null, mock(ScriptManager.class), false);
         usingDocumentFilters = false;
         indexName = UUID.randomUUID().toString();
         timestampField = UUID.randomUUID().toString();

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilderTest.java

@@ -12,6 +12,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.model.codec.JsonFactories;
 
 import java.io.IOException;
 import java.util.Map;
@@ -20,6 +21,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 
 class PullIngestionEnvelopeBuilderTest {
 
@@ -37,7 +39,7 @@ void setUp() {
     }
 
     private PullIngestionEnvelopeBuilder createObjectUnderTest() {
-        return new PullIngestionEnvelopeBuilder(objectMapper);
+        return new PullIngestionEnvelopeBuilder();
     }
 
     @Test
@@ -76,4 +78,19 @@ void build_sets_version_as_string() throws IOException {
         final Map<String, Object> envelope = objectMapper.readValue(result, Map.class);
         assertThat(envelope.get("_version"), equalTo(String.valueOf(version)));
     }
+
+    @Test
+    void build_parses_source_with_string_values_above_the_jackson_default_limit() throws IOException {
+        final int length = 21 * 1024 * 1024;
+        final String sourceJson = "{\"largeField\":\"" + "a".repeat(length) + "\"}";
+
+        final byte[] result = createObjectUnderTest().build(documentId, version, sourceJson, opType);
+
+        @SuppressWarnings("unchecked")
+        final Map<String, Object> envelope = new ObjectMapper(JsonFactories.defaultJsonFactory()).readValue(result, Map.class);
+        assertThat(envelope.get("_source"), instanceOf(Map.class));
+        @SuppressWarnings("unchecked")
+        final Map<String, Object> source = (Map<String, Object>) envelope.get("_source");
+        assertThat(((String) source.get("largeField")).length(), equalTo(length));
+    }
 }