Skip to content

fix jsonToRowTest race condition #36616

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Abacn merged 2 commits into from
Nov 3, 2025
+16 −15
Merged

fix jsonToRowTest race condition #36616

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e630090
fix race condition
derrickaw Oct 24, 2025
fea502d
update based on comments
derrickaw Oct 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 14 additions & 15 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/util/RowJsonUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,21 +51,20 @@ public class RowJsonUtils {
* overwrite the default buffer size limit to 100 MB, and exposes this interface for higher limit.
* If needed, call this method during pipeline run time, e.g. in DoFn.setup.
*/
public static void increaseDefaultStreamReadConstraints(int newLimit) {
if (newLimit <= defaultBufferLimit) {
return;
public static synchronized void increaseDefaultStreamReadConstraints(int newLimit) {
if (newLimit > defaultBufferLimit) {
try {
Class.forName("com.fasterxml.jackson.core.StreamReadConstraints");
Copy link
Contributor

@Abacn Abacn Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we can check STREAM_READ_CONSTRAINTS_AVAILABLE then return early and remove this try {} block


com.fasterxml.jackson.core.StreamReadConstraints.overrideDefaultStreamReadConstraints(
com.fasterxml.jackson.core.StreamReadConstraints.builder()
.maxStringLength(newLimit)
.build());
} catch (ClassNotFoundException e) {
// <2.15, do nothing
}
defaultBufferLimit = newLimit;
}
try {
Class<?> unused = Class.forName("com.fasterxml.jackson.core.StreamReadConstraints");

com.fasterxml.jackson.core.StreamReadConstraints.overrideDefaultStreamReadConstraints(
com.fasterxml.jackson.core.StreamReadConstraints.builder()
.maxStringLength(newLimit)
.build());
} catch (ClassNotFoundException e) {
// <2.15, do nothing
}
defaultBufferLimit = newLimit;
}

static {
Expand Down Expand Up @@ -101,7 +100,7 @@ static void setStreamReadConstraints(JsonFactory jsonFactory, int sizeLimit) {
* factory to higher limits. If needed, call this method during pipeline run time, e.g. in
* DoFn.setup. This avoids a data race caused by modifying the global default settings.
*/
public static JsonFactory createJsonFactory(int sizeLimit) {
public static synchronized JsonFactory createJsonFactory(int sizeLimit) {
Copy link
Contributor

@Abacn Abacn Oct 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shall we confine synchronize {} block only around StreamReadConstraintsHelper.setStreamReadConstraints?

sizeLimit = Math.max(sizeLimit, MAX_STRING_LENGTH);
JsonFactory jsonFactory = new JsonFactory();
if (STREAM_READ_CONSTRAINTS_AVAILABLE) {
Expand Down
Loading