zhipeng93 commented on code in PR #215:
URL: https://github.com/apache/flink-ml/pull/215#discussion_r1168219732
##########
flink-ml-core/src/main/java/org/apache/flink/ml/common/broadcast/operator/AbstractBroadcastWrapperOperator.java:
##########
@@ -169,83 +144,39 @@
parameters.getOperatorEventDispatcher())
.f0;
+ this.indexOfSubtask = containingTask.getIndexInSubtaskGroup();
+ this.numInputs = inTypes.length;
+
boolean hasRichFunction =
wrappedOperator instanceof AbstractUdfStreamOperator
&& ((AbstractUdfStreamOperator)
wrappedOperator).getUserFunction()
instanceof RichFunction;
-
- if (hasRichFunction) {
- wrappedOperatorRuntimeContext =
- new BroadcastStreamingRuntimeContext(
- containingTask.getEnvironment(),
-
containingTask.getEnvironment().getAccumulatorRegistry().getUserMap(),
- wrappedOperator.getMetricGroup(),
- wrappedOperator.getOperatorID(),
- ((AbstractUdfStreamOperator) wrappedOperator)
- .getProcessingTimeService(),
- null,
-
containingTask.getEnvironment().getExternalResourceInfoProvider());
-
- ((RichFunction) ((AbstractUdfStreamOperator)
wrappedOperator).getUserFunction())
- .setRuntimeContext(wrappedOperatorRuntimeContext);
- } else {
- throw new RuntimeException(
- "The operator is not a instance of "
- + AbstractUdfStreamOperator.class.getSimpleName()
- + " that contains a "
- + RichFunction.class.getSimpleName());
- }
-
- this.mailboxExecutor =
-
containingTask.getMailboxExecutorFactory().createExecutor(TaskMailbox.MIN_PRIORITY);
- // variables specific for withBroadcast functionality.
- this.broadcastStreamNames = broadcastStreamNames;
- this.isBlocked = isBlocked;
- this.inTypes = inTypes;
- this.broadcastVariablesReady = false;
- this.indexOfSubtask = containingTask.getIndexInSubtaskGroup();
- this.numInputs = inTypes.length;
-
- // puts in mailboxExecutor
- for (String name : broadcastStreamNames) {
- BroadcastContext.putMailBoxExecutor(name + "-" + indexOfSubtask,
mailboxExecutor);
- }
-
- basePath =
- OperatorUtils.getDataCachePath(
-
containingTask.getEnvironment().getTaskManagerInfo().getConfiguration(),
- containingTask
- .getEnvironment()
- .getIOManager()
- .getSpillingDirectoriesPaths());
- dataCacheWriters = new DataCacheWriter[numInputs];
- hasPendingElements = new boolean[numInputs];
- Arrays.fill(hasPendingElements, true);
+ richContext = new RichContext(hasRichFunction, broadcastStreamNames,
inTypes, isBlocked);
Review Comment:
`RichContext` stores the variables that this operator needs to access the
broadcast variables. If this operator does not need to access broadcast
variables (i.e., has no rich functions), then all the variables in
`RichContext` (except `HasRichFunction`) should be null.
Storing them in `RichContext` improves readbility and also make the
programming easier --- It reminds the developer (me) that I am operating
variables that only operators with richfunction can use these variables.
I can extract it as a separate commit when we get aligned on this PR.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
