diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java index 7e6374ac15..92a8ab6a47 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkIngester.java +++ b/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); diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactory.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactory.java index 60dce2fa8f..ab92042bf9 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactory.java +++ b/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; } diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java index 65d18dee0c..61eabeef96 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java +++ b/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(), diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapper.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapper.java index fa7681cd2c..dbe2424047 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapper.java +++ b/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 { diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverter.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverter.java index 38b0fc44eb..2ab57855e8 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverter.java +++ b/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; diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilder.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilder.java index a340c7c4e8..b0e0576480 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilder.java +++ b/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 { diff --git a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java b/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java index 06e2ab3f07..8355100d2a 100644 --- a/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactory.java +++ b/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) { diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactoryTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactoryTest.java index 789355b0c1..dd79dce9cb 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationFactoryTest.java +++ b/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")); + } } diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java index 61559e7db9..498eb7d7fc 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java +++ b/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); } diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkScriptTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkScriptTest.java index 219f1fba03..517b39e39b 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkScriptTest.java +++ b/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 params) { diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapperTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapperTest.java index 26bae2a1a0..25ecf87223 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/bulk/PreSerializedJsonpMapperTest.java +++ b/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)); } -} \ No newline at end of file + + @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)); + } + } +} diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverterTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverterTest.java index 4020d65a83..bd02ba9898 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/dlq/FailedBulkOperationConverterTest.java +++ b/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 doc = (Map) ((FailedDlqData) result.getFailedData()).getDocument(); + assertThat(((String) doc.get("largeField")).length(), is(equalTo(length))); + } } diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/TimeSeriesIndexTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/TimeSeriesIndexTest.java index 2927bbb715..c29c944098 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/TimeSeriesIndexTest.java +++ b/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(); diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilderTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilderTest.java index b2e573cb30..aebaa034f9 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull_ingestion/PullIngestionEnvelopeBuilderTest.java +++ b/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 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 envelope = new ObjectMapper(JsonFactories.defaultJsonFactory()).readValue(result, Map.class); + assertThat(envelope.get("_source"), instanceOf(Map.class)); + @SuppressWarnings("unchecked") + final Map source = (Map) envelope.get("_source"); + assertThat(((String) source.get("largeField")).length(), equalTo(length)); + } } diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryJsonpMapperTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryJsonpMapperTest.java new file mode 100644 index 0000000000..cb84978f35 --- /dev/null +++ b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryJsonpMapperTest.java @@ -0,0 +1,76 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.dataprepper.plugins.source.opensearch.worker.client; + +import com.fasterxml.jackson.core.exc.StreamConstraintsException; +import com.fasterxml.jackson.databind.node.ObjectNode; +import jakarta.json.JsonException; +import jakarta.json.stream.JsonParser; +import org.junit.jupiter.api.Test; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; +import org.opensearch.dataprepper.model.codec.JsonFactories; + +import java.io.StringReader; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class OpenSearchClientFactoryJsonpMapperTest { + + @Test + void default_JacksonJsonpMapper_rejects_string_values_over_the_jackson_default_limit() { + final String json = jsonWithStringValueOfLength(20_000_001); + + final JsonException exception = assertThrows(JsonException.class, + () -> deserialize(new JacksonJsonpMapper(), json)); + + assertThat(exception.getCause(), instanceOf(StreamConstraintsException.class)); + } + + @Test + void createJacksonJsonpMapper_deserializes_string_values_above_the_jackson_default_limit() { + final int length = 32 * 1024 * 1024; + final String json = jsonWithStringValueOfLength(length); + + final ObjectNode result = deserialize(OpenSearchClientFactory.createJacksonJsonpMapper(), json); + + assertThat(result.get("largeField").textValue().length(), equalTo(length)); + } + + @Test + void createJacksonJsonpMapper_rejects_string_values_above_the_raised_limit() { + final String json = jsonWithStringValueOfLength(JsonFactories.MAX_JSON_STRING_LENGTH + 1); + + final JsonException exception = assertThrows(JsonException.class, + () -> deserialize(OpenSearchClientFactory.createJacksonJsonpMapper(), json)); + + assertThat(exception.getCause(), instanceOf(StreamConstraintsException.class)); + } + + @Test + void createElasticsearchJacksonJsonpMapper_raises_the_max_string_length() { + final co.elastic.clients.json.jackson.JacksonJsonpMapper mapper = OpenSearchClientFactory.createElasticsearchJacksonJsonpMapper(); + + assertThat(mapper.objectMapper().getFactory().streamReadConstraints().getMaxStringLength(), + equalTo(JsonFactories.MAX_JSON_STRING_LENGTH)); + } + + private static ObjectNode deserialize(final JacksonJsonpMapper mapper, final String json) { + try (final JsonParser parser = mapper.jsonProvider().createParser(new StringReader(json))) { + return mapper.deserialize(parser, ObjectNode.class); + } + } + + private static String jsonWithStringValueOfLength(final int length) { + return "{\"largeField\":\"" + "a".repeat(length) + "\"}"; + } +} diff --git a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryTest.java b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryTest.java index ecc7d54787..91941a67b9 100644 --- a/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryTest.java +++ b/data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/client/OpenSearchClientFactoryTest.java @@ -14,9 +14,12 @@ import org.mockito.Mock; import org.mockito.MockedStatic; import org.mockito.junit.jupiter.MockitoExtension; +import org.opensearch.client.json.JsonpMapper; +import org.opensearch.client.json.jackson.JacksonJsonpMapper; import org.opensearch.client.opensearch.OpenSearchClient; import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions; import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier; +import org.opensearch.dataprepper.model.codec.JsonFactories; import org.opensearch.dataprepper.plugins.source.opensearch.AuthConfig; import org.opensearch.dataprepper.plugins.source.opensearch.OpenSearchSourceConfiguration; import org.opensearch.dataprepper.plugins.source.opensearch.configuration.AwsAuthenticationConfiguration; @@ -37,6 +40,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.notNullValue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.lenient; @@ -113,6 +117,7 @@ void provideOpenSearchClient_with_deprecated_username_and_password() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); verifyNoInteractions(awsCredentialsSupplier); @@ -135,6 +140,7 @@ void provideOpenSearchClient_with_username_and_password() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); verifyNoInteractions(awsCredentialsSupplier); @@ -154,6 +160,7 @@ void provideElasticSearchClient_with_deprecated_username_and_password() { final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); verifyNoInteractions(awsCredentialsSupplier); } @@ -173,6 +180,7 @@ void provideElasticSearchClient_with_username_and_password() { final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); verifyNoInteractions(awsCredentialsSupplier); } @@ -196,6 +204,7 @@ void provideElasticSearchClient_with_aws_auth() { final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); final AwsCredentialsOptions awsCredentialsOptions = awsCredentialsOptionsArgumentCaptor.getValue(); assertThat(awsCredentialsOptions, notNullValue()); @@ -223,6 +232,7 @@ void provideOpenSearchClient_with_aws_auth() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); final AwsCredentialsOptions awsCredentialsOptions = awsCredentialsOptionsArgumentCaptor.getValue(); assertThat(awsCredentialsOptions, notNullValue()); @@ -242,6 +252,7 @@ void provideElasticSearchClient_with_auth_disabled() { final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); verifyNoInteractions(awsCredentialsSupplier); verify(openSearchSourceConfiguration, never()).getUsername(); @@ -259,6 +270,7 @@ void provideOpenSearchClient_with_auth_disabled() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); verifyNoInteractions(awsCredentialsSupplier); verify(openSearchSourceConfiguration, never()).getUsername(); @@ -284,6 +296,7 @@ void provideOpenSearchClient_with_aws_auth_and_serverless_flag_true() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); final AwsCredentialsOptions awsCredentialsOptions = awsCredentialsOptionsArgumentCaptor.getValue(); assertThat(awsCredentialsOptions, notNullValue()); @@ -307,6 +320,7 @@ void provideOpenSearchClient_with_deprecated_self_signed_certificate() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(path)); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); } } @@ -325,6 +339,7 @@ void provideOpenSearchClient_with_self_signed_certificate_filepath() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(certificatePathCaptor.capture())); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); assertThat(certificatePathCaptor.getValue(), equalTo(Path.of(certificate))); } } @@ -343,6 +358,7 @@ void provideOpenSearchClient_with_self_signed_certificate_content() { final OpenSearchClient openSearchClient = createObjectUnderTest().provideOpenSearchClient(openSearchSourceConfiguration); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(TEST_CERTIFICATE)); assertThat(openSearchClient, notNullValue()); + assertMaxStringLengthIsRaised(openSearchClient); } } @@ -360,6 +376,7 @@ void provideElasticSearchClient_with_deprecated_self_signed_certificate_filepath .thenReturn(sslContext); final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(path)); } } @@ -378,6 +395,7 @@ void provideElasticSearchClient_with_self_signed_certificate_filepath() { .thenReturn(sslContext); final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(certificatePathCaptor.capture())); assertThat(certificatePathCaptor.getValue().toString(), equalTo(certificate)); } @@ -396,6 +414,7 @@ void provideElasticSearchClient_with_self_signed_certificate_content() { .thenReturn(sslContext); final ElasticsearchClient elasticsearchClient = createObjectUnderTest().provideElasticSearchClient(openSearchSourceConfiguration); assertThat(elasticsearchClient, notNullValue()); + assertMaxStringLengthIsRaised(elasticsearchClient); trustStoreProviderMockedStatic.verify(() -> TrustStoreProvider.createSSLContext(TEST_CERTIFICATE)); } } @@ -448,4 +467,17 @@ void createSdkAsyncHttpClient_with_insecure_configuration_and_no_cert_path_trust } } + private static void assertMaxStringLengthIsRaised(final OpenSearchClient openSearchClient) { + final JsonpMapper jsonpMapper = openSearchClient._transport().jsonpMapper(); + assertThat(jsonpMapper, instanceOf(JacksonJsonpMapper.class)); + assertThat(((JacksonJsonpMapper) jsonpMapper).objectMapper().getFactory().streamReadConstraints().getMaxStringLength(), + equalTo(JsonFactories.MAX_JSON_STRING_LENGTH)); + } + + private static void assertMaxStringLengthIsRaised(final ElasticsearchClient elasticsearchClient) { + final co.elastic.clients.json.JsonpMapper jsonpMapper = elasticsearchClient._transport().jsonpMapper(); + assertThat(jsonpMapper, instanceOf(co.elastic.clients.json.jackson.JacksonJsonpMapper.class)); + assertThat(((co.elastic.clients.json.jackson.JacksonJsonpMapper) jsonpMapper).objectMapper().getFactory().streamReadConstraints().getMaxStringLength(), + equalTo(JsonFactories.MAX_JSON_STRING_LENGTH)); + } }