pnowojski commented on a change in pull request #11177: [FLINK-16219][runtime]
Made AsyncWaitOperator chainable to non-sources.
URL: https://github.com/apache/flink/pull/11177#discussion_r383141152
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/ChainingStrategy.java
##########
@@ -38,16 +38,46 @@
* <p>To optimize performance, it is generally a good practice to allow
maximal
* chaining and increase operator parallelism.
*/
- ALWAYS,
+ ALWAYS {
+ @Override
+ public boolean canBeChainedTo(StreamOperatorFactory<?>
headOperator) {
+ return true;
+ }
+ },
/**
* The operator will not be chained to the preceding or succeeding
operators.
*/
- NEVER,
+ NEVER {
+ @Override
+ public boolean canBeChainedTo(StreamOperatorFactory<?>
headOperator) {
+ return false;
+ }
+ },
/**
* The operator will not be chained to the predecessor, but successors
may chain to this
* operator.
*/
- HEAD
+ HEAD {
+ @Override
+ public boolean canBeChainedTo(StreamOperatorFactory<?>
headOperator) {
+ return false;
+ }
+ },
+
+ /**
+ * Operators will be eagerly chained whenever possible, except after
legacy sources.
+ *
+ * <p>Operators that will not properly when processInput is called from
another thread, must use this strategy
+ * instead of {@link #ALWAYS}.
+ */
+ HEAD_AFTER_LEGACY_SOURCE {
+ @Override
+ public boolean canBeChainedTo(StreamOperatorFactory<?>
headOperator) {
+ return !headOperator.isStreamSource();
+ }
+ };
+
+ public abstract boolean canBeChainedTo(StreamOperatorFactory<?>
headOperator);
Review comment:
Is it working properly? What if we have `StreamSource -> OP2 -> OP3 ->
AsyncWaitOperator`?
----------------------------------------------------------------
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]
With regards,
Apache Git Services
