New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[FLINK-11084]Throw a hard exception to remind developers while there's no stream node between two split transformation #7258
Conversation
e36be96
to
6438c8d
Compare
cc @twalthr |
cc @aljoscha |
…s no stream node between two split transformation
43915c1
to
1350ace
Compare
Thanks for the PR @Clarkkkkk . based on the JIRA discussion, should we also put a I fully agree consecutive split are buggy features within proper safeguarding and also can be done much more elegantly using side output. Any other concerns that we want to keep the |
@walterddr Good point, I will tag it. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think generally looks good. Had just a concern about applying split to side-output. Could you check if that combination works?
|
||
private <T> void validateSplitTransformation(StreamTransformation<T> input) { | ||
if (input instanceof SelectTransformation || input instanceof SplitTransformation) { | ||
throw new IllegalStateException("Error while tranforming SplitTransformation, please use side output instead."); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you change the exception message to sth along the lines: Consecutive multiple splits are not supported. Splits are deprecated. Please use side-outputs.
@@ -643,4 +645,18 @@ private String determineSlotSharingGroup(String specifiedGroup, Collection<Integ | |||
return inputGroup == null ? "default" : inputGroup; | |||
} | |||
} | |||
|
|||
private <T> void validateSplitTransformation(StreamTransformation<T> input) { | |||
if (input instanceof SelectTransformation || input instanceof SplitTransformation) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What about SideOutputTransformation
? Can we apply split on top of side-output?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@dawidwys Good point.
|
||
DataStreamSource<Integer> src = env.fromElements(0, 0); | ||
|
||
OutputSelector<Integer> outputSelector = new OutputSelector<Integer>() { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could you extract common outputSelector
?
cc @dawidwys |
…s no stream node between two split transformation This closes apache#7258
…s no stream node between two split transformation This closes apache#7258
…s no stream node between two split transformation This closes apache#7258
+1, merging when my travis gives green |
…mations This closes apache#7258
What is the purpose of the change
This PR throws a hard exception to remind developers while there's no stream node between two split transformation and recommend developers to use side output instead of split stream.
Brief change log
Verifying this change
This change added tests and can be verified as follows:
DataStreamTest::testConsecutiveSplitRejection
DataStreamTest::testSelectBetweenConsecutiveSplitRejection
DataStreamTest::testUnionBetweenConsecutiveSplitRejection
DataStreamTest::testKeybyBetweenConsecutiveSplitRejection
Does this pull request potentially affect one of the following parts:
@Public(Evolving)
: noDocumentation