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:
[email protected]
