Skip to content

[PubsubIO] Account for message metadata in bounded write batch size - #39886

Open
1fanwang wants to merge 1 commit into
apache:masterfrom
1fanwang:pubsub-bounded-batch-size-28011
Open

[PubsubIO] Account for message metadata in bounded write batch size#39886
1fanwang wants to merge 1 commit into
apache:masterfrom
1fanwang:pubsub-bounded-batch-size-28011

Conversation

@1fanwang

Copy link
Copy Markdown

Bounded PubsubIO writes size each batch from the message payload alone. Attributes and ordering keys are omitted, so the writer can publish a batch larger than maxPublishBatchByteSize.

validatePubsubMessage already returns the validated size of the payload, ordering key, and attributes. The bounded writer now uses that value instead of discarding it and recalculating messageSize from the payload.

This does not address #31800. Its existing TODOs cover JSON versus Protobuf encoding and HTTP to gRPC transcoding.

Fixes #28011


Testing

The regression sends two 27-byte messages through the real bounded-writer lifecycle with a 40-byte batch limit and captures each publish call. To reproduce against base 13875fc6bd3d5ad534c89d33f2395176fbc485fc while keeping the regression test from this branch:

git checkout 13875fc6bd3d5ad534c89d33f2395176fbc485fc -- \
  sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
export JAVA_HOME=$(/usr/libexec/java_home -v 21.0.6)
export PATH="$JAVA_HOME/bin:$PATH"
./gradlew :sdks:java:io:google-cloud-platform:test \
  --tests org.apache.beam.sdk.io.gcp.pubsub.PubsubIOTest.testBoundedWriteBatchSizeIncludesAttributesAndOrderingKey \
  --no-daemon --console=plain
git checkout HEAD -- \
  sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

Both messages were sent in one batch:

java.lang.AssertionError: expected:<[1, 1]> but was:<[2]>
1 test completed, 1 failed
BUILD FAILED in 17s

After the fix:

export JAVA_HOME=$(/usr/libexec/java_home -v 21.0.6)
export PATH="$JAVA_HOME/bin:$PATH"
./gradlew :sdks:java:io:google-cloud-platform:test \
  --tests org.apache.beam.sdk.io.gcp.pubsub.PubsubIOTest.testBoundedWriteBatchSizeIncludesAttributesAndOrderingKey \
  --no-daemon --console=plain
BUILD SUCCESSFUL in 33s

The captured batch sizes are now [1, 1], so neither publish exceeds the configured limit. The full PubsubIOTest class also passes with the same Gradle task and the class-level test selector.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
@github-actions

Copy link
Copy Markdown
Contributor

Assigning reviewers:

R: @kennknowles for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

  • stop reviewer notifications - opt out of the automated review tooling
  • remind me after tests pass - tag the comment author after tests pass
  • waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug]: PubsubIO used in batch incorrect batch cutoff size

1 participant