Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/6087
Thanks for opening the PR @zhangminglei. I agree with @sihuazhou that it's
not as easy as replacing `x` with `xAsync`. As @sihuazhou pointed out, we have
to react to asynchronous Yarn messages depending on our internal state. So for
example, when calling `startContainerAsync` we should store somewhere that we
tried to start a container. If this request later fails, we should not fail the
complete ResourceManager but rather send a new container request (as it is done
in the `YarnResourceManager#onContainersAllocated` method).
Additionally, while touching this code, we should check whether we can
improve the unit tests for this component. Especially with the asynchronous
node manager client, it should be rather simple to write good tests when
creating a mock implementation and manually calling the callbacks.
---
