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

[1fanwang]

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.

• This PR fixes [Bug]: PubsubIO used in batch incorrect batch cutoff size #28011.
• CHANGES.md is unchanged for this small bug fix.
• This is not a large contribution, so the ICLA checklist item does not apply.

[github-actions]

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).

[github-actions]

Reminder, please take a look at this pr: @kennknowles

[github-actions]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @ahmedabu98 for label java.

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)

[kennknowles]

R: @sjvanrossum

Since you made some of the surrounding comments, are you able to review this?

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers