[FLINK-14214] Performance regression in TwoInputStreamOperator

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Cannot Reproduce
• Affects Version/s: 1.10.0
• Component/s: Runtime / Task
• Labels: pull-request-available
• Assignee: Piotr Nowojski
• Reporter: Piotr Nowojski
• Created: 25/Sep/19 15:17
• Updated: 02/Oct/19 11:45
• Resolved: 02/Oct/19 11:45

Description

https://issues.apache.org/jira/browse/FLINK-13051 introduced a performance regression visible for example here as a drop on August 30th (note that the later performance improvement is caused by serialisation improvement, which is unrelated to the previous slow down).

Probable suspect is the following .isDone() check executed inside StreamTwoInputProcessor once per record:

// to avoid starvation, if the input selection is ALL and availableInputsMask is not ALL,
// always try to check and set the availability of another input
// TODO: because this can be a costly operation (checking volatile inside CompletableFuture`
//  this might be optimized to only check once per processed NetworkBuffer
if (inputSelectionHandler.shouldSetAvailableForAnotherInput()) {
	checkAndSetAvailable(1 - readingInputIndex);
}

Issue Links

• is caused by: FLINK-13051 Drop the non-selectable two-input StreamTask and Processor
• is related to: FLINK-11859 Improve SpanningRecordSerializer performance by serializing record length to serialization buffer directly
• links to: GitHub Pull Request #9779