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

[slilichenko]

What happened?

This is due to incorrect initialization of a transform [1] - instead of the max batch size in bytes the max number of batch records is passed. Addition issues with the message validation: Error message here [2] is misleading. It should state that a single record size exceeds maximum batch size, rather than just the number of bytes referenced in the generic Pub/Sub limits.

[1]

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

Line 1363 in e59f001

new PubsubBoundedWriter(

[2]

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PreparePubsubWriteDoFn.java

Line 100 in e59f001

if (totalSize > maxPublishBatchSize) {

Issue Priority

Priority: 3 (minor)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[liferoad]

is this same as #27000?

[slilichenko]

Yes, I was using 2.48.0 and validated that things work after upgrading to 2.49.0.

Technically, there is still a bug - total message size including attribute data returned by the "validate" method is ignored and instead the message data size is used to calculate the batch cutoff:

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

Line 1441 in 369265d

PreparePubsubWriteDoFn.validatePubsubMessageSize(message, maxPublishBatchByteSize);

[Abacn]

Per the latest comment, changed the Issue title and put proper priority tag