scwhittle commented on code in PR #38454:
URL: https://github.com/apache/beam/pull/38454#discussion_r3316239433
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupAlsoByWindowsParDoFn.java:
##########
@@ -142,6 +142,13 @@ public void processTimers() throws Exception {
// it here to build a KeyedWorkItem
}
+ @Override
+ public void finishKey() throws Exception {
+ if (fnRunner != null) {
Review Comment:
should we assert it is not null instead?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java:
##########
@@ -363,6 +363,13 @@ public void processTimers() throws Exception {
processTimers(TimerType.SYSTEM, stepContext, windowCoder);
}
+ @Override
+ public void finishKey() throws Exception {
+ if (fnRunner != null) {
Review Comment:
should we check is not null instead? it seems the other methods assert that
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunner.java:
##########
@@ -63,6 +63,12 @@ public interface DoFnRunner<InputT extends @Nullable Object,
OutputT extends @Nu
<KeyT extends @Nullable Object> void onWindowExpiration(
BoundedWindow window, Instant timestamp, KeyT key);
+ /**
+ * Performs per-key cleanup or processing after all elements and timers for
a key have been
Review Comment:
should we say something like before transitioning to a new key or finishing
the bundle? I'm wondering if it is possible for a runner driving this to
process a batch of input for a key, then a different key, and then back to the
original key.
Otherwise I think it also should be clearer I think that "all elements and
timers" are scoped to those within the bundle.
--
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]
