flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From GitBox <...@apache.org>
Subject [GitHub] [flink] rkhachatryan commented on a change in pull request #10151: [FLINK-14231] Handle the pending processing-time timers to make endInput semantics on the operator chain strict
Date Thu, 16 Jan 2020 08:34:08 GMT
rkhachatryan commented on a change in pull request #10151: [FLINK-14231] Handle the pending
processing-time timers to make endInput semantics on the operator chain strict
URL: https://github.com/apache/flink/pull/10151#discussion_r367288070
 
 

 ##########
 File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
 ##########
 @@ -140,34 +140,41 @@ public OperatorChain(
 			}
 
 			// we create the chain of operators and grab the collector that leads into the chain
-			List<StreamOperator<?>> allOps = new ArrayList<>(chainedConfigs.size());
+			List<StreamOperatorWrapper<?, ?>> allOpWrappers = new ArrayList<>(chainedConfigs.size());
 			this.chainEntryPoint = createOutputCollector(
 				containingTask,
 				configuration,
 				chainedConfigs,
 				userCodeClassloader,
 				streamOutputMap,
-				allOps,
+				allOpWrappers,
 				containingTask.getMailboxExecutorFactory());
 
 			if (operatorFactory != null) {
 				WatermarkGaugeExposingOutput<StreamRecord<OUT>> output = getChainEntryPoint();
 
-				headOperator = StreamOperatorFactoryUtil.createOperator(
+				OP headOperator = StreamOperatorFactoryUtil.createOperator(
 						operatorFactory,
 						containingTask,
 						configuration,
 						output);
 
 				headOperator.getMetricGroup().gauge(MetricNames.IO_CURRENT_OUTPUT_WATERMARK, output.getWatermarkGauge());
+				this.headOperatorWrapper = createOperatorWrapper(headOperator, containingTask, configuration);
+
+				// add head operator to end of chain
+				allOpWrappers.add(headOperatorWrapper);
+
+				this.tailOperatorWrapper = allOpWrappers.get(0);
 			} else {
-				headOperator = null;
+				checkState(allOpWrappers.size() == 0);
 
 Review comment:
   What I see from code, is that it depends solely on configuration:
   ```
   final StreamConfig configuration = containingTask.getConfiguration();
   ```
   For `allOpWrappers`:
   ```
   in constructor:
   this.chainEntryPoint = createOutputCollector(containingTask, configuration, ...
   
   in createOutputCollector:
       for (StreamEdge outputEdge : operatorConfig.getChainedOutputs(userCodeClassloader))
{
           WatermarkGaugeExposingOutput<StreamRecord<T>> output = createChainedOperator(...
allOperatorWrappers ...)
   
   in createChainedOperator:
       allOperatorWrappers.add(createOperatorWrapper(chainedOperator, containingTask, operatorConfig));
   ```
   
   For `headOperatorWrapper`:
   ```
   in constructor:
   StreamOperatorFactory<OUT> operatorFactory = configuration.getStreamOperatorFactory(userCodeClassloader);
   if (operatorFactory != null) {
       ...
   } else {
       checkState(allOpWrappers.size() == 0);
       this.headOperatorWrapper = null;
   ```
   
   So it'd be safer to remove this check and
   if `allOpWrappers` is not empty then assign `headOperatorWrapper` to the last element.

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

Mime
View raw message