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 needs to access broadcast
variables (i.e., has no rich functions), then all the variables in
`RichContext` (except `HasRichFunction`) should be null.
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
