scwhittle commented on code in PR #39961:
URL: https://github.com/apache/beam/pull/39961#discussion_r4046269359
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java:
##########
@@ -717,21 +717,25 @@ private void flushStateInternal() {
// If activeReader is null, we might still have backlogBytes from an
SDF. We ignore a reported
// backlogBytes of 1 since older versions of the Java SDK use this value
as a default when
// RestrictionTracker.getProgress() or GetSize() are not defined.
- getOutputBuilder().setSourceBacklogBytes(backlogBytes);
+ getKeyOutputBuilder().setSourceBacklogBytes(backlogBytes);
}
- getOutputBuilder()
+ getKeyOutputBuilder()
.setSourceBytesProcessed(computeSourceBytesProcessed(sourceBytesProcessCounterName));
validateCommitRequestSize();
- getOrCreateWorkItemCommits().add(getOutputBuilder().build());
- this.outputBuilder = null;
+ if (this.workItemCommits == null) {
+ this.workItemCommits =
Review Comment:
I was thinking this could avoid ArrayList without multikey bundles.
if (multikeyBundleOptions.multiKeyBundleEnabled()) {
if (workItemCommits == null) workItemCommits = new ArrayList();
workItemCommits.add(...);
} else {
checkState(workItemCommits == null);
workItemCommits = ImmutableList.of(getKeyOutputBuilder().build());
}
--
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]
