arunpandianp commented on code in PR #39961:
URL: https://github.com/apache/beam/pull/39961#discussion_r3965486773
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java:
##########
@@ -350,26 +350,47 @@ public long add(WindowedValue<T> data) throws IOException
{
return (long) key.size() + value.size() + metadata.size() + id.size() +
offsetSize;
}
- @Override
- public void close() throws IOException {
+ private void flush(boolean bundleLevel) {
try {
outputBuilder.setDestinationStreamId(destinationName);
for (Windmill.KeyedMessageBundle.Builder keyedOutput :
productionMap.values()) {
outputBuilder.addBundles(keyedOutput.build());
}
if (outputBuilder.getBundlesCount() > 0) {
- context.getOutputBuilder().addOutputMessages(outputBuilder.build());
+ Windmill.OutputMessageBundle bundle = outputBuilder.build();
+ if (bundleLevel) {
+ context.addBundleOutputMessages(bundle);
Review Comment:
In WindmillSink, attaching the outputs with the keys that produced them
might be useful since doing so will preserve the key-key ordering as long as
there are no buffered outputs. I'm thinking to keep it without cross key
merging, till we figure out the story around ordering. wdyt?
--
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]
