[FLINK-19255] Add configuration to make AsyncWaitOperation Chainable

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Information Provided
• Affects Version/s: 1.10.2, 1.11.2
• Component/s: API / Core

Environment

Any flink job using Async IO post this PR: https://github.com/apache/flink/pull/11177/files#diff-00b95aec1fa804a531b553610ec74c27R117

(so I believe anything starting at either 1.9 or 1.10).

 

• Assignee: Unassigned
• Reporter: Kyle Bendickson
• Created: 16/Sep/20 05:05
• Updated: 22/Jun/21 13:55
• Resolved: 27/Sep/20 06:21

Description

Currently, we no longer chain Async IO calls. Instead, anything using AsyncIO starts the new head of an operator chain as a temporary workaround for this issue: https://issues.apache.org/jira/browse/FLINK-13063

 

However, because this change can (and does in my customers' cases) have very large impact on the job graph size, and because people were previously accepting of their results, in the 1.10 release it was made so that AsyncWaitOperator could be chained in this issue https://issues.apache.org/jira/browse/FLINK-16219.

 

However, it's very complicated and not intuitive for users to call out to operator factory methods. I have users who would very much like to not have their AsyncIO calls generate a new chain, as it's ballooned the number of state stores they have and they were accepting of their previous results. The only exmaple I could find was in the tests, and its rather convoluted.

 

My proposal would be to add that config check just before the line of code in AsyncWaitOperator.java that would not add the following line, which is currently hardcoded into the operator and what requires one to use the operator factory:

setChainingStrategy(ChainingStrategy.HEAD)

  

Given that this is considered potentially unsafe / legacy behavior, I would suggest that we add a config, something that explicitly calls this out as unsafe / legacy, so that users do not have to go through the unintuitive process of using operator factories but that new users know not to use this option or to use it at their own risk. We could also document that it is not necessarily going to be supported in the future if need be.

 

My suggestion for config names that would avoid that setChainingStrategy line include

taskmanager.async-io-operator.legacy-unsafe-chaining-strategy

which specifically calls this behavior out as legacy and unsafe.

 

Another possible name could be

pipeline.operator-chaining.async-io.legacy-inconsistent-chaining-strategy-always

(which would be more in line with the existing config of pipeline.operator-chaining).

 

 

Given that it is possible to stop operator chaining, it's just very unintuitive and requires using operator factories, I think that this configuration would be a good addition. I would be happy to submit a PR, with tests, and updated documentation, so that power users who are looking to do this could enable / disable this behavior without having to change their code much.

 

I recognize that this might be an odd request as this has been deemed unsafe, but this change has made it very difficult for some of my users to use rocksdb, namely those with very large state that previously made very liberal use of Async IO (especially for things like analytics events which can be sent on a best effort basis) and who therefore have a very large job graph after this change.

 

If anybody has any better suggestions for names, I'd be open to them. And then as mentioned, I'd be happy to submit a PR with tests etc.

 

For reference, here are the tests where I found the ability to use the operator factory and here is the utility function which is needed to create a chained async io operator vertex. Note that this utility function is in the test and not part of the public facing API. https://github.com/apache/flink/blob/3a04e179e09224b09c4ee656d31558844da83a26/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java#L880-L912

If there is a simpler way to handle this, I'd be happy to hear it. Otherwise, since this behavior is technically already specifically enabled (as called out in the changelog from Flink 1.11), I think it makes sense to add a config and either document that its legacy behavior, unsafe (or inconsistent, up to you), and that it could go away at any time.

 

But it seems unnecessary to require users to go through so many extra hoops in the code, especially for users who share operators amongst different jobs which might be configured to use different state backends. Not to mention that some of these users want the legacy behavior and others would prefer to play it safe and accept the additional shuffle, so a code fix is not always feasible when code is shared, but the enabling / disabling of a cluster level config would still allow for shared code.

 

I'd be happy to submit a patch once this issue is discussed.

 

Thank you,

Kyle B. - Data Services @ Tinder 

Issue Links

• duplicates: FLINK-16219 Make AsyncWaitOperator chainable again