[GitHub] azagrebin commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead…

[GitBox <gi...@apache.org>]

azagrebin commented on a change in pull request #7255: [FLINK-10945] Use InputDependencyConstraint to avoid resource dead…
URL: https://github.com/apache/flink/pull/7255#discussion_r240313527
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
 ##########
 @@ -726,6 +728,56 @@ void sendPartitionInfos() {
 		}
 	}
 
+	/**
+	 * Check whether the InputDependencyConstraint is satisfied for this vertex.
+	 *
+	 * @return whether the input constraint is satisfied
+	 */
+	public boolean checkInputDependencyConstraints() {
 
 Review comment:
   This might be a bit more concise:
   ```
      boolean checkInputDependencyConstraints() {
   		if (getExecutionGraph().getInputDependencyConstraint() == InputDependencyConstraint.ANY) {
   			// InputDependencyConstraint == ANY
   			return jobVertex.getInputs().stream().anyMatch(this::isInputConsumable);
   		} else {
   			// InputDependencyConstraint == ALL
   			return jobVertex.getInputs().stream().allMatch(this::isInputConsumable);
   		}
   	}
   
   	boolean isInputConsumable(IntermediateResult result) {
   		if (result.getResultType().isPipelined()) {
   			// For PIPELINED result, the input is consumable if any result partition has produced records or is finished
   			return Arrays.stream(result.getPartitions()).anyMatch(IntermediateResultPartition::hasDataProduced);
   		} else {
   			// For BLOCKING result, the input is consumable if all the partitions in the result are finished
   			return result.areAllPartitionsFinished();
   		}
   	}
   ```
   I am not sure we need to check the `ANY` case at all. Just checking this theoretically changes current behaviour. On the other hand, at the moment, I think it is always true for `ANY` where we check it if `ScheduleMode.LAZY_FROM_SOURCES`. 
   
   I am also not sure that `ALL` config makes sense together with `ScheduleMode.EAGER`.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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