[FLINK-8816][ResourceManager] Remove the oldWorker only after starting newWorker successfully

[sihuazhou]

What is the purpose of the change

This PR address FLINK-8816, We should remove the oldWorker only after starting newWorker successfully in ResourceManager#registerTaskExecutorInternal and javadoc for SlotProvider#allocateSlot is outdated.

Brief change log

• Fix bug registerTaskExecutorInternal(), remove old worker after starting newWorker successfully.
• Fix javadoc for SlotProvider#allocateSlot.

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes)
• The S3 file system connector: (no)

Documentation

no

[sihuazhou]

Hi, @tillrohrmann Could you please have a look at this?

[tillrohrmann]

Why do you want to keep a TaskExecutor which does has not been started by the ResourceManager? That way we could end up using resources which don't belong to our Yarn application, for example.

[sihuazhou]

@tillrohrmann I just doubt whether the new TaskExecutor is a faker, If it's a faker, it will evict the old valid one. I'd like to close this If you think this can't be happen ;).

[tillrohrmann]

I think this cannot happen since a TaskExecutor will only try to reconnect if the previous connection attempt failed/timed out. So there should not be the chance that you receive two registration success messages if I'm not mistaken.

[sihuazhou]

Aha, I almost like to close this PR now. About the javadoc outdated, I think it can be fixed in some other PR, it just a trivial work ;).