This closes #9141

[FLINK-13241][yarn/mesos] Fix Yarn/MesosResourceManager setting managed memory size into wrong configuration instance.

[FLINK-13241][yarn][test] Update YarnResourceManagerTest#testCreateSlotsPerWorker to compute tmCalculatedResourceProfile based on the RM altered configuration.

[FLINK-13241][yarn][test] Update YarnConfigurationITCase to verify that TMs are started with correct managed memory size.

[FLINK-13241][runtime] Calculating and set managed memory size outside of ResourceManager.

[FLINK-13241][rumtime/yarn][test] Move YarnResourceManagerTest#testCreateSlotsPerWorker to ResourceManagerTest#testCreateWorkerSlotProfiles, and update to verify slot profile calculation with determinate managed memory size.

[FLINK-13241][runtime] Move getResourceManagerConfiguration from ResourceManagerFactory to ResourceManagerUtil.

This closes #9246.

Only kill Yarn application if it does not properly terminate.

This closes #9175.

This closes #9229.

[hotfix][python] Change "flink-python-" to "flink-python" for the change  of artifact of flink-python module (#9270)

This closes #9203

This closes #9227.

This closes #9249

Currently test cases will fail when trying to close the output stream if all data written
but ClosedByInterruptException occurs at the ending phase. This commit fixes it.

This closes #9235

This closes #9257.

On producer side the netty handler receives the CancelPartitionRequest for releasing the SubpartitionView resource.
In previous implementation we try to find the corresponding view via available queue in PartitionRequestQueue. But
in reality the view is not always available to stay in this queue, then the view would never be released.

Furthermore the release of ResultPartition/ResultSubpartitions is based on the reference counter in ReleaseOnConsumptionResultPartition,
but while handling the CancelPartitionRequest in PartitionRequestQueue, the ReleaseOnConsumptionResultPartition is never
notified of consumed subpartition. That means the reference counter would never decrease to 0 to trigger partition release,
which would bring file resource leak in the case of BoundedBlockingSubpartition.

In order to fix above two issues, the corresponding view is released via all reader queue instead, and then it would call
ReleaseOnConsumptionResultPartition#onConsumedSubpartition meanwhile to solve this bug.

This closes #9241.

The method Unsafe.defineClass() is removed in Java 11. To support Java 11, we rework the method
"CommonTestUtils.createClassNotInClassPath()" to use a different mechanism.

This commit now writes the class byte code out to a temporary file and create a new URLClassLoader that
loads the class from that file.  That solution is not a complete drop-in replacement, because it cannot
add the class to an existing class loader, but can only create a new pair of (classloader & new-class-in-that-classloader).
Because of that, the commit also adjusts the existing tests to work with that new mechanism.

This closes #9251

This closes #9088

This closes #9223

FLINK-13249 was a bug where a deadlock occurred when the network thread got blocked on a lock
while requesting partitions to be read by remote channels. The test mimicks that situation
to guard the fix applied in an earlier commit.