respond to comments

Author: rwiggles

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

@@ -723,13 +723,8 @@ private void validateCommitRequestSize() {
     // so, we're purposefully dropping them here
     Windmill.WorkItemCommitRequest.Builder truncationBuilder =
         buildWorkItemTruncationRequestBuilder(currentWork, estimatedCommitSize);
-    for (int i = 0; i < outputBuilders.size(); i++) {
-      if (outputBuilders.get(i) == currentBuilder) {
-        outputBuilders.set(i, truncationBuilder);
-        break;
-      }
-    }
-    this.outputBuilder = truncationBuilder;
+    this.outputBuilder.clear();
+    this.outputBuilder.mergeFrom(truncationBuilder);
   }
 
   private Windmill.WorkItemCommitRequest.Builder buildWorkItemTruncationRequestBuilder(

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/streaming/KeyCommitTooLargeException.java

@@ -22,19 +22,6 @@
 
 public final class KeyCommitTooLargeException extends Exception {
 
-  public static KeyCommitTooLargeException causedBy(
-      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);
-  }
-
   public static KeyCommitTooLargeException causedBy(
       String stageName,
       long byteLimit,

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java

@@ -1401,15 +1401,18 @@ private void runKeyCommitTooLargeExceptionTest(
                 1, "large_key", DEFAULT_SHARDING_KEY, largeCommit.getEstimatedWorkItemCommitBytes())
             .build(),
         removeDynamicFields(largeCommit));
-
+    // Check this explicitly since the estimated commit bytes weren't actuallyExpand commentComment on line L1340
+    // checked against an expected value in the previous step
     assertTrue(largeCommit.getEstimatedWorkItemCommitBytes() > 1000);
 
+    // Spam worker updates a few times.
     int maxTries = 10;
     while (--maxTries > 0) {
       worker.reportPeriodicWorkerUpdatesForTest();
       Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
     }
 
+    // We should see an exception reported for the large commit but not the small one.
     ArgumentCaptor<WorkItemStatus> workItemStatusCaptor =
         ArgumentCaptor.forClass(WorkItemStatus.class);
     verify(mockWorkUnitClient, atLeast(2)).reportWorkItemStatus(workItemStatusCaptor.capture());