acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4342989905
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to
report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
Here were some alternatives provided by AI, though it's first recommendation
is what I didn't really want to do:
1. Add a capability method to ParDoFn (Recommended)
Add default boolean isSplittable() { return false; } (or
shouldReportBytesRead()) on ParDoFn (and delegate in ForwardingParDoFn).
Determine it via the actual DoFn class rather than any string:
SplittableProcessFnFactory.ProcessFnExtractor wraps the user DoFn in
SplittableParDoViaKeyedWorkItems.ProcessFn, which is already accessible at
construction time via helpers.doFnInstanceManager.peek().getDoFn() instanceof
ProcessFn inside SimpleParDoFnHelpers (used by both SimpleParDoFn and
StreamingKeyedWorkItemSideInputParDoFn).
Alternatively, SplittableProcessFnFactory.createDefault() (or
UserParDoFnFactory.streamingKeyedWorkItem, which is already true exclusively
for SplittableProcessFnFactory) can set/wrap that flag on the returned ParDoFn.
IntrinsicMapTaskExecutorFactory then simply checks if (fn.isSplittable()),
keeping all CloudObject dispatch encapsulated inside DefaultParDoFnFactory.
2. Have SplittableProcessFnFactory wrap with a dedicated ForwardingParDoFn
subclass (or marker interface)
If you don't want to touch the public ParDoFn interface,
SplittableProcessFnFactory can wrap the ParDoFn produced by UserParDoFnFactory
in a package-private SplittableParDoFn extends ForwardingParDoFn (or marker
interface), allowing fn instanceof SplittableParDoFn in
IntrinsicMapTaskExecutorFactory.
3. Share a constant if keeping the CloudObject check
If you prefer keeping the check on cloudUserFn.getClassName(), extracting a
shared constant (e.g., SplittableProcessFnFactory.OBJECT_TYPE_NAME =
"SplittableProcessFn") used by DefaultParDoFnFactory,
IntrinsicMapTaskExecutorFactory, and tests avoids duplicating the string
literal (similar to ComputationWorkExecutorFactory.isCustomSource), though
Option 1 or 2 is cleaner since it delegates the classification to
ParDoFnFactory.
--
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]
