pnowojski commented on a change in pull request #11098: [FLINK-16060][task]
Implement working StreamMultipleInputProcessor
URL: https://github.com/apache/flink/pull/11098#discussion_r382674662
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InputSelection.java
##########
@@ -110,8 +101,39 @@ public int fairSelectNextIndexOutOf2(int
availableInputsMask, int lastReadInputI
throw new UnsupportedOperationException("Only two inputs are
supported.");
}
- private static boolean isALLMaskOf2(long inputMask) {
- return (3 & inputMask) == 3;
+ /**
+ * Fairly select one of the available inputs for reading.
+ *
+ * @param availableInputsMask The mask of all available inputs.
+ * @param lastReadInputIndex The index of last read input.
+ * @return the index of the input for reading or -1, and -1 indicates
no input is selected (
+ * {@code inputMask} is empty or the inputs in {@code
inputMask} are unavailable).
+ */
+ public int fairSelectNextIndex(int availableInputsMask, int
lastReadInputIndex) {
+ int selectionMask = (int) inputMask;
+ int combineMask = availableInputsMask & selectionMask;
Review comment:
Good question, I had to double check that. `availableInputsMask` could only
be set to `-1` to indicate all 32 inputs are available. I will document that.
`MultipleInputSelectionHandler` is adhering to this contract.
----------------------------------------------------------------
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
