arunpandianp commented on code in PR #39316:
URL: https://github.com/apache/beam/pull/39316#discussion_r3633974644
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -1336,8 +1337,8 @@ public void testKeyCommitTooLargeException() throws
Exception {
1, "large_key", DEFAULT_SHARDING_KEY,
largeCommit.getEstimatedWorkItemCommitBytes())
.build(),
removeDynamicFields(largeCommit));
-
- // Check this explicitly since the estimated commit bytes weren't actually
+ // Check this explicitly since the estimated commit bytes weren't
actuallyExpand commentComment
+ // on line L1340
Review Comment:
fix comment
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -669,6 +681,47 @@ private void flushStateInternal() {
getOutputBuilder()
.setSourceBytesProcessed(computeSourceBytesProcessed(sourceBytesProcessCounterName));
+
+ validateCommitRequestSize();
+ }
+
+ private void validateCommitRequestSize() {
+ Windmill.WorkItemCommitRequest.Builder currentBuilder = getOutputBuilder();
+ Work currentWork = getWork();
+ long byteLimit = operationalLimits.getMaxWorkItemCommitBytes();
+ Windmill.WorkItemCommitRequest commitRequest = currentBuilder.build();
+ int commitSize = commitRequest.getSerializedSize();
+ int estimatedCommitSize = commitSize < 0 ? Integer.MAX_VALUE : commitSize;
+
+ // Detect overflow of integer serialized size or if the byte limit was
exceeded.
+ // Commit is too large if overflow has occurred or the commitSize has
exceeded the allowed
+ // commit byte limit.
Review Comment:
```suggestion
// Detect overflow of integer serialized size or if the byte limit was
exceeded.
// Commit is too large if overflow has occurred or the commitSize has
exceeded the allowed
// commit byte limit.
int estimatedCommitSize = commitSize < 0 ? Integer.MAX_VALUE :
commitSize;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
