Skip to content

Commit 1aba88b

Browse files
dlvenabledlvenable
committed
Consolidates the code for configuring 64 MB when deserializing from JSON.
Adds a new JsonFactories class that creates a default JsonFactory. Use this in code where we updated to 64 MB already and in the ndjson and json codecs. Signed-off-by: David Venable <dlv@amazon.com>
1 parent b7f400d commit 1aba88b

11 files changed

Lines changed: 245 additions & 32 deletions

File tree

‎data-prepper-api/build.gradle‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,11 @@ dependencies {
2828
}
2929

3030

31+
test {
32+
// Tests verifying the raised maximum string length build strings larger than 64 MB.
33+
maxHeapSize = '2g'
34+
}
35+
3136
tasks.named('validateCompatibility') {
3237
methodExcludes.add('org.opensearch.dataprepper.model.annotations.Experimental#message()')
3338
methodExcludes.add('org.opensearch.dataprepper.model.annotations.Experimental#groups()')

‎data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
import com.fasterxml.jackson.core.JsonFactory;
1313
import com.fasterxml.jackson.core.JsonParser;
1414
import com.fasterxml.jackson.core.JsonToken;
15-
import com.fasterxml.jackson.core.StreamReadConstraints;
1615
import com.fasterxml.jackson.databind.ObjectMapper;
1716

1817
import org.opensearch.dataprepper.model.event.Event;
@@ -29,8 +28,8 @@
2928
import java.util.function.Consumer;
3029

3130
public class JsonDecoder implements ByteDecoder {
32-
private final ObjectMapper objectMapper = new ObjectMapper();
33-
private final JsonFactory jsonFactory = new JsonFactory();
31+
private final ObjectMapper objectMapper;
32+
private final JsonFactory jsonFactory;
3433
private String keyName;
3534
private Collection<String> includeKeys;
3635
private Collection<String> includeKeysMetadata;
@@ -39,17 +38,17 @@ public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<St
3938
this.keyName = keyName;
4039
this.includeKeys = includeKeys;
4140
this.includeKeysMetadata = includeKeysMetadata;
42-
if (maxEventLength != null) {
43-
jsonFactory.setStreamReadConstraints(StreamReadConstraints.builder()
44-
.maxStringLength(maxEventLength)
45-
.build());
46-
}
41+
this.jsonFactory = maxEventLength != null ?
42+
JsonFactories.jsonFactory(maxEventLength) : JsonFactories.defaultJsonFactory();
43+
this.objectMapper = new ObjectMapper(jsonFactory);
4744
}
4845

4946
public JsonDecoder() {
5047
this.keyName = null;
5148
this.includeKeys = null;
5249
this.includeKeysMetadata = null;
50+
this.jsonFactory = JsonFactories.defaultJsonFactory();
51+
this.objectMapper = new ObjectMapper(jsonFactory);
5352
}
5453

5554
public void parse(InputStream inputStream, Instant timeReceived, Consumer<Record<Event>> eventConsumer) throws IOException {

‎data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonFactories.java‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.model.codec;
11+
12+
import com.fasterxml.jackson.core.JsonFactory;
13+
import com.fasterxml.jackson.core.StreamReadConstraints;
14+
15+
/**
16+
* Provides {@link JsonFactory} instances configured with Data Prepper's stream read constraints.
17+
* <p>
18+
* Events may hold individual string values larger than Jackson's default maximum string length
19+
* of 20 MB. Any code which parses JSON into events, or which re-reads JSON that Data Prepper
20+
* itself wrote (such as codecs reading from a buffer), should create its {@link JsonFactory}
21+
* from this class so that the same maximum string length applies consistently.
22+
*
23+
* @since 2.17
24+
*/
25+
public final class JsonFactories {
26+
/**
27+
* The maximum JSON string length Data Prepper supports in an event: 64 MB.
28+
*/
29+
public static final int MAX_JSON_STRING_LENGTH = 64 * 1024 * 1024;
30+
31+
private JsonFactories() {
32+
}
33+
34+
/**
35+
* Creates a {@link JsonFactory} with the default Data Prepper maximum string length.
36+
*
37+
* @return a new {@link JsonFactory}
38+
* @since 2.17
39+
*/
40+
public static JsonFactory defaultJsonFactory() {
41+
return jsonFactory(MAX_JSON_STRING_LENGTH);
42+
}
43+
44+
/**
45+
* Creates a {@link JsonFactory} with the provided maximum string length.
46+
*
47+
* @param maxStringLength the maximum JSON string length the factory's parsers permit
48+
* @return a new {@link JsonFactory}
49+
* @since 2.17
50+
*/
51+
public static JsonFactory jsonFactory(final int maxStringLength) {
52+
return JsonFactory.builder()
53+
.streamReadConstraints(StreamReadConstraints.builder()
54+
.maxStringLength(maxStringLength)
55+
.build())
56+
.build();
57+
}
58+
}

‎data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java‎

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,8 @@
99

1010
package org.opensearch.dataprepper.model.event;
1111

12-
import com.fasterxml.jackson.core.JsonFactory;
1312
import com.fasterxml.jackson.core.JsonPointer;
1413
import com.fasterxml.jackson.core.JsonProcessingException;
15-
import com.fasterxml.jackson.core.StreamReadConstraints;
1614
import com.fasterxml.jackson.core.type.TypeReference;
1715
import com.fasterxml.jackson.databind.JsonNode;
1816
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -26,6 +24,7 @@
2624
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
2725
import org.apache.commons.lang3.StringUtils;
2826
import org.opensearch.dataprepper.expression.ExpressionEvaluator;
27+
import org.opensearch.dataprepper.model.codec.JsonFactories;
2928
import org.opensearch.dataprepper.model.event.exceptions.EventKeyNotFoundException;
3029
import org.slf4j.Logger;
3130
import org.slf4j.LoggerFactory;
@@ -101,15 +100,7 @@ public DefaultEventFailureMetadata withErrorMessage(Object value) {
101100

102101
private static final String SEPARATOR = "/";
103102

104-
private static final int MAX_JSON_STRING_LENGTH = 64 * 1024 * 1024;
105-
106-
private static final JsonFactory JSON_FACTORY = JsonFactory.builder()
107-
.streamReadConstraints(StreamReadConstraints.builder()
108-
.maxStringLength(MAX_JSON_STRING_LENGTH)
109-
.build())
110-
.build();
111-
112-
private static final ObjectMapper mapper = JsonMapper.builder(JSON_FACTORY)
103+
private static final ObjectMapper mapper = JsonMapper.builder(JsonFactories.defaultJsonFactory())
113104
.disable(JsonNodeFeature.STRIP_TRAILING_BIGDECIMAL_ZEROES)
114105
.build()
115106
.registerModule(new JavaTimeModule())

‎data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonDecoderTest.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,26 @@ void test_basicJsonDecoder_withMaxEventLength() {
109109
assertThat(map.get("key1"), equalTo(validString));
110110
}
111111

112+
@Test
113+
void test_basicJsonDecoder_withNullMaxEventLength_usesDefaultMaximumStringLength() {
114+
// With no max_event_length configured, the decoder uses the Data Prepper default of
115+
// 64 MB, which is larger than Jackson's default of 20 MB.
116+
String largeString = "x".repeat(21 * 1024 * 1024);
117+
String inputString = "[{\"key1\":\"" + largeString + "\"}]";
118+
119+
jsonDecoder = new JsonDecoder(null, null, null, null);
120+
121+
assertDoesNotThrow(() -> {
122+
jsonDecoder.parse(new ByteArrayInputStream(inputString.getBytes()), null, (record) -> {
123+
receivedRecord = record;
124+
});
125+
});
126+
127+
assertNotNull(receivedRecord);
128+
Map<String, Object> map = receivedRecord.getData().toMap();
129+
assertThat(map.get("key1"), equalTo(largeString));
130+
}
131+
112132
@Test
113133
void test_basicJsonDecoder_withTimeReceived() {
114134
String stringValue = UUID.randomUUID().toString();

‎data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonFactoriesTest.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.model.codec;
11+
12+
import com.fasterxml.jackson.core.JsonFactory;
13+
import org.junit.jupiter.api.Test;
14+
import org.junit.jupiter.params.ParameterizedTest;
15+
import org.junit.jupiter.params.provider.ValueSource;
16+
17+
import static org.hamcrest.CoreMatchers.equalTo;
18+
import static org.hamcrest.CoreMatchers.not;
19+
import static org.hamcrest.CoreMatchers.sameInstance;
20+
import static org.hamcrest.MatcherAssert.assertThat;
21+
22+
class JsonFactoriesTest {
23+
@Test
24+
void defaultJsonFactory_returns_factory_with_the_data_prepper_maximum_string_length() {
25+
final JsonFactory jsonFactory = JsonFactories.defaultJsonFactory();
26+
27+
assertThat(jsonFactory.streamReadConstraints().getMaxStringLength(),
28+
equalTo(JsonFactories.MAX_JSON_STRING_LENGTH));
29+
}
30+
31+
@Test
32+
void defaultJsonFactory_returns_a_new_factory_for_each_call() {
33+
assertThat(JsonFactories.defaultJsonFactory(), not(sameInstance(JsonFactories.defaultJsonFactory())));
34+
}
35+
36+
@ParameterizedTest
37+
@ValueSource(ints = {1024, 20_000_000, 64 * 1024 * 1024})
38+
void jsonFactory_returns_factory_with_the_provided_maximum_string_length(final int maxStringLength) {
39+
final JsonFactory jsonFactory = JsonFactories.jsonFactory(maxStringLength);
40+
41+
assertThat(jsonFactory.streamReadConstraints().getMaxStringLength(), equalTo(maxStringLength));
42+
}
43+
}

‎data-prepper-plugins/event-json-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodec.java‎

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,14 +12,14 @@
1212
import com.fasterxml.jackson.core.JsonFactory;
1313
import com.fasterxml.jackson.core.JsonParser;
1414
import com.fasterxml.jackson.core.JsonToken;
15-
import com.fasterxml.jackson.core.StreamReadConstraints;
1615
import com.fasterxml.jackson.databind.ObjectMapper;
1716
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
1817

1918
import org.opensearch.dataprepper.model.configuration.DataPrepperVersion;
2019
import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
2120
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
2221
import org.opensearch.dataprepper.model.codec.InputCodec;
22+
import org.opensearch.dataprepper.model.codec.JsonFactories;
2323
import org.opensearch.dataprepper.model.event.Event;
2424
import org.opensearch.dataprepper.model.event.EventType;
2525
import org.opensearch.dataprepper.model.event.EventMetadata;
@@ -45,16 +45,7 @@
4545
public class EventJsonInputCodec implements InputCodec {
4646
private static final Logger LOG = LoggerFactory.getLogger(JacksonEvent.class);
4747

48-
// Events may hold individual string values larger than Jackson's default maximum string
49-
// length of 20 MB. Use the same 64 MB limit as JacksonEvent so that any event which was
50-
// written by the EventJsonOutputCodec can be read back.
51-
static final int MAX_JSON_STRING_LENGTH = 64 * 1024 * 1024;
52-
53-
private final JsonFactory jsonFactory = JsonFactory.builder()
54-
.streamReadConstraints(StreamReadConstraints.builder()
55-
.maxStringLength(MAX_JSON_STRING_LENGTH)
56-
.build())
57-
.build();
48+
private final JsonFactory jsonFactory = JsonFactories.defaultJsonFactory();
5849
private final ObjectMapper objectMapper = new ObjectMapper(jsonFactory).registerModule(new JavaTimeModule());
5950
private final Boolean overrideTimeReceived;
6051

‎data-prepper-plugins/parse-json-processor/build.gradle‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
48
*/
59

610
plugins {
@@ -23,6 +27,8 @@ dependencies {
2327

2428
test {
2529
useJUnitPlatform()
30+
// Tests verifying the raised maximum string length build strings larger than 64 MB.
31+
maxHeapSize = '2g'
2632
}
2733

2834
jacocoTestCoverageVerification {

‎data-prepper-plugins/parse-json-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/NdjsonInputCodec.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
48
*/
59

610
package org.opensearch.dataprepper.plugins.codec.json;
@@ -14,6 +18,7 @@
1418
import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
1519
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
1620
import org.opensearch.dataprepper.model.codec.InputCodec;
21+
import org.opensearch.dataprepper.model.codec.JsonFactories;
1722
import org.opensearch.dataprepper.model.event.Event;
1823
import org.opensearch.dataprepper.model.event.EventFactory;
1924
import org.opensearch.dataprepper.model.event.LogEventBuilder;
@@ -33,7 +38,7 @@
3338
@DataPrepperPlugin(name = "ndjson", pluginType = InputCodec.class, pluginConfigurationType = NdjsonInputConfig.class)
3439
public class NdjsonInputCodec implements InputCodec {
3540
private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE = new TypeReference<>() {};
36-
private final ObjectMapper objectMapper = new ObjectMapper();
41+
private final ObjectMapper objectMapper;
3742
private final NdjsonInputConfig ndjsonInputConfig;
3843
private final EventFactory eventFactory;
3944
private final JsonFactory jsonFactory;
@@ -42,7 +47,8 @@ public class NdjsonInputCodec implements InputCodec {
4247
public NdjsonInputCodec(final NdjsonInputConfig ndjsonInputConfig, final EventFactory eventFactory) {
4348
this.ndjsonInputConfig = ndjsonInputConfig;
4449
this.eventFactory = eventFactory;
45-
jsonFactory = new JsonFactory();
50+
jsonFactory = JsonFactories.defaultJsonFactory();
51+
objectMapper = new ObjectMapper(jsonFactory);
4652
}
4753

4854
@Override

‎data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/JsonInputCodecTest.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,16 @@
11
/*
22
* Copyright OpenSearch Contributors
33
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
48
*/
59

610
package org.opensearch.dataprepper.plugins.codec.json;
711

812
import com.fasterxml.jackson.core.JsonProcessingException;
13+
import com.fasterxml.jackson.core.exc.StreamConstraintsException;
914
import com.fasterxml.jackson.databind.ObjectMapper;
1015
import org.junit.jupiter.api.BeforeEach;
1116
import org.junit.jupiter.api.Test;
@@ -489,4 +494,42 @@ private void parse_InputStream_withEventConfig(final int numberOfObjects, final
489494
assertThat(actualRecord.getData().getMetadata().getEventType(), equalTo(EventType.LOG.toString()));
490495
}
491496
}
497+
498+
@ParameterizedTest
499+
@ValueSource(ints = {32, 64})
500+
void parse_with_string_value_within_the_raised_limit_returns_event(final int valueSizeInMB) throws IOException {
501+
// Jackson's default maximum string length is 20 MB; the codec raises it to 64 MB.
502+
// String values up to that limit must parse successfully.
503+
final String key = UUID.randomUUID().toString();
504+
final String largeValue = buildStringOfSize(valueSizeInMB * 1024 * 1024);
505+
final InputStream inputStream = createInputStream(
506+
Map.of("events", List.of(Map.of(key, largeValue))));
507+
508+
createObjectUnderTest().parse(inputStream, eventConsumer);
509+
510+
final ArgumentCaptor<Record<Event>> recordArgumentCaptor = ArgumentCaptor.forClass(Record.class);
511+
verify(eventConsumer).accept(recordArgumentCaptor.capture());
512+
assertThat(recordArgumentCaptor.getValue().getData().get(key, String.class), equalTo(largeValue));
513+
}
514+
515+
@Test
516+
void parse_with_string_value_exceeding_the_raised_limit_throws() throws IOException {
517+
// A string value larger than the 64 MB limit must still be rejected.
518+
final String key = UUID.randomUUID().toString();
519+
final String tooLargeValue = buildStringOfSize(65 * 1024 * 1024);
520+
final InputStream inputStream = createInputStream(
521+
Map.of("events", List.of(Map.of(key, tooLargeValue))));
522+
523+
assertThrows(StreamConstraintsException.class,
524+
() -> createObjectUnderTest().parse(inputStream, eventConsumer));
525+
verifyNoInteractions(eventConsumer);
526+
}
527+
528+
private static String buildStringOfSize(final int length) {
529+
final StringBuilder builder = new StringBuilder(length);
530+
while (builder.length() < length) {
531+
builder.append('a');
532+
}
533+
return builder.toString();
534+
}
492535
}

0 commit comments

Comments
 (0)