[GitHub] [flink] aljoscha commented on a change in pull request #13795: [FLINK-19814][FLINK-19815] Do not support iterations in BATCH mode.

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

aljoscha commented on a change in pull request #13795:
URL: https://github.com/apache/flink/pull/13795#discussion_r512562273



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
##########
@@ -517,6 +523,12 @@ private boolean isUnboundedSource(final Transformation<?> transformation) {
 	 */
 	private <F> Collection<Integer> transformCoFeedback(CoFeedbackTransformation<F> coIterate) {
 
+		if (shouldExecuteInBatchMode) {
+			throw new UnsupportedOperationException("Iterations are not supported in BATCH" +
+					" execution mode. If you want to execute such a pipeline, please set the " +
+					"'" + ExecutionOptions.RUNTIME_MODE.key() + "'=STREAMING");

Review comment:
       Same as above.

##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java
##########
@@ -434,6 +434,12 @@ private boolean isUnboundedSource(final Transformation<?> transformation) {
 	 */
 	private <T> Collection<Integer> transformFeedback(FeedbackTransformation<T> iterate) {
 
+		if (shouldExecuteInBatchMode) {
+			throw new UnsupportedOperationException("Iterations are not supported in BATCH" +
+					" execution mode. If you want to execute such a pipeline, please set the " +
+					"'" + ExecutionOptions.RUNTIME_MODE.key() + "'=STREAMING");

Review comment:
       Should also use the enum value instead of the hardcoded `"STREAMING"` string.

##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorBatchExecutionTest.java
##########
@@ -381,6 +394,44 @@ public void testInputSelectableMultiInputTransformation() {
 		graphGenerator.generate();
 	}
 
+	@Test
+	public void testFeedbackThrowsExceptionInBatch() {
+		final SourceTransformation<Integer> bounded =
+				new SourceTransformation<>(
+						"Bounded Source",
+						new SourceOperatorFactory<>(new MockSource(Boundedness.BOUNDED, 100), WatermarkStrategy.noWatermarks()),
+						IntegerTypeInfo.of(Integer.class),
+						1);
+
+		final FeedbackTransformation<Integer> feedbackTransformation =
+				new FeedbackTransformation<>(bounded, 5L);
+
+		testNoSupportForIterationsInBatchHelper(bounded, feedbackTransformation);
+	}
+
+	@Test
+	public void testCoFeedbackThrowsExceptionInBatch() {
+		final CoFeedbackTransformation<Integer> coFeedbackTransformation =
+				new CoFeedbackTransformation<>(2, TypeInformation.of(Integer.TYPE), 5L);
+		testNoSupportForIterationsInBatchHelper(coFeedbackTransformation);
+	}
+
+	private void testNoSupportForIterationsInBatchHelper(final Transformation<?>... transformations) {
+		final List<Transformation<?>> registeredTransformations = new ArrayList<>();
+		Collections.addAll(registeredTransformations, transformations);
+
+		StreamGraphGenerator streamGraphGenerator =
+				new StreamGraphGenerator(
+						registeredTransformations,
+						new ExecutionConfig(),
+						new CheckpointConfig());
+		streamGraphGenerator.setRuntimeExecutionMode(RuntimeExecutionMode.AUTOMATIC);

Review comment:
       Why don't you set it to `BATCH` to be more specific in the test?




----------------------------------------------------------------
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