arunpandianp commented on code in PR #39316:
URL: https://github.com/apache/beam/pull/39316#discussion_r3611109924
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -1337,18 +1338,14 @@ public void testKeyCommitTooLargeException() throws
Exception {
.build(),
removeDynamicFields(largeCommit));
- // Check this explicitly since the estimated commit bytes weren't actually
Review Comment:
keep the comments?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -669,6 +681,52 @@ 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.
+
streamingCounters.windmillMaxObservedWorkItemCommitBytes().addValue(estimatedCommitSize);
+ if (commitSize >= 0 && commitSize < byteLimit) {
+ return;
+ }
+
+ KeyCommitTooLargeException e =
+ KeyCommitTooLargeException.causedBy(
+ systemName, byteLimit, commitRequest, key, hotKeyLoggingEnabled);
+ failureTracker.trackFailure(systemName, currentWork.getWorkItem(), e);
+ LOG.error("{}", e.toString());
+
+ // Drop the current request in favor of a new, minimal one requesting
truncation.
+ // Messages, timers, counters, and other commit content will not be used
by the service
+ // so, we're purposefully dropping them here
+ Windmill.WorkItemCommitRequest.Builder truncationBuilder =
+ buildWorkItemTruncationRequestBuilder(currentWork,
estimatedCommitSize);
+ for (int i = 0; i < outputBuilders.size(); i++) {
Review Comment:
You can do
```
this.outputBuilder.clear();
this.outputBuilder.mergeFrom(truncationBuilder);
```
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/KeyCommitTooLargeException.java:
##########
@@ -18,16 +18,38 @@
package org.apache.beam.runners.dataflow.worker.streaming;
import org.apache.beam.runners.dataflow.worker.windmill.Windmill;
+import org.checkerframework.checker.nullness.qual.Nullable;
public final class KeyCommitTooLargeException extends Exception {
public static KeyCommitTooLargeException causedBy(
- String computationId, long byteLimit, Windmill.WorkItemCommitRequest
request) {
+ String stageName, long byteLimit, Windmill.WorkItemCommitRequest
request) {
+ return causedBy(stageName, byteLimit, request, null, false);
+ }
+
+ public static KeyCommitTooLargeException causedBy(
+ String stageName,
+ long byteLimit,
+ Windmill.WorkItemCommitRequest request,
+ boolean hotKeyLoggingEnabled) {
+ return causedBy(stageName, byteLimit, request, null, hotKeyLoggingEnabled);
+ }
Review Comment:
these look unused, remove?
--
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]
