scwhittle commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4398449549
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/ParDoFn.java:
##########
@@ -37,4 +38,8 @@ public interface ParDoFn {
void finishBundle() throws Exception;
void abort() throws Exception;
+
+ default @Nullable DoFnInfo<?, ?> getDoFnInfo() throws Exception {
+ return null;
Review Comment:
perhaps it would be better to remove the default so that new classes (and
perhaps other existing ones) return an appropriate dofninfo?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingKeyedWorkItemSideInputParDoFn.java:
##########
@@ -231,18 +226,13 @@ protected void
onProcessWindowedValue(WindowedValue<KeyedWorkItem<K, InputT>> el
}
if (hasState) {
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>) getDoFnInfo().getWindowingStrategy(),
windowsProcessed);
+ (WindowingStrategy<?, W>) helpers.fnInfo.getWindowingStrategy(),
windowsProcessed);
}
}
- /**
- * Returns the {@link DoFnInfo} currently being used by this {@link
SimpleParDoFn}.
- *
- * <p>May be null if no element has been processed yet, or if the {@link
SimpleParDoFn} has
- * finished.
- */
- @VisibleForTesting
- @Nullable DoFnInfo<?, ?> getDoFnInfo() {
- return helpers.fnInfo;
+ /** Returns the {@link DoFnInfo} used by this {@link
StreamingKeyedWorkItemSideInputParDoFn}. */
Review Comment:
hmm, it appears these DoFnInfo are stateful in that they contain a DoFn
which goes through the dofn lifecycle. previously we would return only the
active dofninfo (or null if not processing). Now we always return one but it
might not be an active one or one that is then activated in another thread.
We don't want callers of getDoFnInfo to call methods that expect to just be
used from the active thread (like dofn lifecycle etc). It seems like it would
be cleanest if all the immutable stuff within DoFnInfo was a different
interface which was returned so that getDoFnInfo couldn't be misused.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java:
##########
@@ -112,7 +111,7 @@ protected void onStartKey() {
// These elements are now processed. Register cleanup timers
for all the unblocked
// windows.
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>)
getDoFnInfo().getWindowingStrategy(),
+ (WindowingStrategy<?, W>)
helpers.fnInfo.getWindowingStrategy(),
Review Comment:
maybe we should rename helpers fnInfo to activeFnInfo or something
indicating it is the fninfo on the active thread?
See comment below
--
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]
