Rename _configuration to originalConfiguration

Remove testing classes from main scope in flink-runtime

Previously, the ForkableFlinkMiniCluster which resided in flink-test-utils required
these files to be in the main scope of flink-runtime. With the removal of the
ForkableFlinkMiniCluster, these classes are now no longer needed and can be moved
back to the test scope.

This closes #2450.

This closes #2478

This closes #2402

Adds a NoOpOperator which is unwound in OperatorTranslation.translate.
This will be first used by Gelly as a placeholder to support implicit
operator reuse.

This closes #2294

Return a DefaultAWSCredentialsProviderChain instead of null when
AWS_CREDENTIALS_PROVIDER config is set to "AUTO"

This closes #2470

Yarn reports null or (1, maxVcores) depending on its internal logic. The
test only worked in the past because it summed up the used vcores of the
RM and the TM containers. We have checks in place to ensure the vcores
config value is passed on to the Flink ResourceManager.

There is RocksDBAsyncSnapshotTest which tests async snapshots for the
RocksDB state backend. Operators themselves cannot do asynchronous
checkpoints right now.

This closes #2424

This caused Scalastyle to fail, presumably depending on the locale
used. After a bit of debugging on the Scalastyle plugin I found out that
the number in the error is the byte position.

"Expected identifier, but got Token(COMMA,,,1772,,)"

head -c 1772 flink-mesos/src/test/scala/org/apache/flink/mesos/Utils.scala

pointed to the Unicode character '⇒' which causes Scalastyle to fail in
certain environments.

This closes #2466

This closes #2441.

[FLINK-4525] [core] (followup) Remove remaining redundant code for pre-defined strictly local assignments.

[FLINK-4490] [distributed coordination] (part 3) Rename methods on 'Instance' to have more intuitive names

getResourceID() --> getTaskManagerID()
getInstanceConnectionInfo() --> getTaskManagerLocation()

To allow for a future dynamic slot allocation and release model, the slots should not depend on 'Instance'.
In this change, the Slots hold most of the necessary information directly (location, gateway) and
the interact with the Instance only via a 'SlotOwner' interface.

[FLINK-4490] [distributed coordination] (part 1) Change InstanceConnectionInfo to TaskManagerLocation

This adds the ResourceId to the TaskManagerLocation

The version change didn't cause the Scalastyle errors. Seems like the
only viable solution to prevent random failures of the Scalastyle
plugin is to disable Scalastyle checks for the affected source file.

[FLINK-4455] [FLINK-4424] [networkenv] Make NetworkEnvironment independent of ActorGateway and JobManager association

Makes the NetworkEnvironment independent of the JobManager association. This means that the
NetworkEnvironment and with it the ConnectionManager is started before the TaskManager actor
is executed. Furthermore, the ConnectionManager keeps running even in case of a JobManager
disassocation. In the wake of the remodelling this behaviour, the PartitionStateChecker and
the ResultPartitionConsumableNotifier which depend on the JobManager association were moved
out of the NetworkEnvironment. They are now contained in the SlotEnvironment which will be
set up when the TaskManager connects to a JobManager. The SlotEnvironment contains all
information related to the associated JobManager. Since all slots are implicitly associated
with the JobManager which is the leader, we only create one SlotEnvironment which is shared
by all Tasks.

Introduce SlotEnvironment to accommodate the PartitionStateChecker and ResultPartitionConsumableNotifier

Remove the PartitionStateChecker and the ResultPartitionConsumableNotifier from the
NetworkEnvironment. Start the NetworkEnvironment when the TaskManager components are
created. Keep the NetworkEnvironment running also when the JobManager is disassociated.

Fix CassandraConnectorITCase

Remove ExecutionContext from TaskManager; Rename SlotEnvironment into JobManagerConnection

Introduce JobManagerCommunicationFactory to generate job manager specific communication components

This closes #2449.

This closes #2462

This closes #2446

This doesn't work yet because the state query machinery is not yet
properly aware of key-grouped state.

This change makes the RocksDB backend key-group aware by building on the
changes in the previous commit.

The biggest change in this is that functionality that used to be in
AbstractStateBackend is now moved to CheckpointStreamFactory and
KeyedStateBackend. The former is responsible for providing streams that
can be used to checkpoint data while the latter is responsible for
keeping keyed state. A keyed backend can checkpoint the state that it
keeps by using a CheckpointStreamFactory.

This also refactors how asynchronous keyed state snapshots work. They
are not implemented using a Future/RunnableFuture.

Also, this changes the keyed state backends to be key-group aware and to
snapshot the state in key-groups with an index for restoring.

This tests the rescaling features in CheckpointCoordinator and
SavepointCoordinator.

This introduces a new KeySelector that assigns keys to key groups and
also adds the max parallelism parameter throughout all API levels.

This also adds tests for the newly introduced features.

The test now properly uses and reuses a mini cluster, rather than spawning a local environment for each test.
This also properly renames the CassandraConnectorTest to CassandraConnectorITCase

[FLINK-4539] [runtime] Reuse functionality for Physical Memory size in 'Hardware' and 'EnvironmentInformation'.

This closes #2432

This closes #2305