acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4327857012
##########
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:
My first thought was to add something more generically to ParDo, but I feel
that this is more of a Dataflow implementation detail that we even need to do
this, so I'd rather not add something to the generic API to specify which
ParDo's should be reporting their bytes to autoscaling. Maybe there's some way
to add something to the CloudObject in SplittableProcessFnFactory.java?
--
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]
