The HandlerRedirectUtils.getRedirectAddress decides whether the retrieved leader address is equal to the local job manager address. The local job manager address is, however, in the form akka.tcp://flink@url/user/jobmanager whereas the leader address can be akka://flink/user/jobmanager if the local job manager is the current leader. Such a case produced a warning which is not correct. This PR checks for the local job manager address and signals that no redirection has to be done if it receives akka://flink/user/jobmanager.

Add test for HandlerRedirectUtils

This closes #1280.

This was previously reported as FLINK-2625 (commit 8a849372).
The Client class was concurrently refactored with FLINK-2097 (commit 71bf2f57).

This closes #1278

A collection of small documentation and grammar updates.

This closes #1277

This closes #1275

Multiple TaskManager filesystems can be used by configuring multiple temporary directories.
This patch changes the process of spilling files from a per-operator round-robin to a global
round-robin such that each directory is written to in turn across all operators, reducing
unbalanced I/O due to bunching.

This closes #1272

This closes #1260

This closes #1170

This adds Time.milliseconds(), Time.seconds(), and so on.

[FLINK-2872] [Documentation] Update the documentation for Scala part to add ExecutionEnvironment.readFileOfPrimitives.

Add the missing Scala part for ExecutionEnvironment.readFileOfPrimitives API doc in the
programming guide.

Author: Henry Saputra <hsaputra@apache.org>

Closes #1268 from hsaputra/add_readFileOfPrimitives_scala_guide and squashes the following commits:

f543436 [Henry Saputra] Add variant of ExecutionEnvironment.readFileOfPrimitives in Java with additional delimiter.
1209019 [Henry Saputra] [FLINK-2872] [Documentation] Update the documentation for Scala part to add readFileOfPrimitives.

This also renames WatermarkTrigger to EventTimeTrigger and
ContinuousWatermarkTrigger to ContinuousEventTimeTrigger.

This adds method state() on Trigger context that should be used to
create an OperatorState to deal with fault-tolerant state.

WindowAssigner now has a method getWindowSerializer() that is used to
get a TypeSerializer for the Windows that it assigns. The Serializer for
the Key is retrieved from the input KeyedStream and the serializer for
the input elements is already available.

During checkpointing all currently in-flight windows (per key, per
window) are serialized using the TypeSerializers. The state that is
accessible in Triggers using state() is kept in a
HashMap<String, Serializable>, this is serialized using java
serialization.

- set the upper bound to Long.MAX_VALUE

For YARN, we set it to the calculated maximum container size (no need to fix).

[FLINK-2354] [runtime] Replace old StateHandleProvider by StateStorageHelper in ZooKeeperStateHandleStore

The old StateHandleProvider used in ZooKeeperStateHandleStore had to be replaced because the state backend implementation has changed. Since the new state backend could not be used anymore, a new StateStorageHelper interface has been created. The default implementation FileSystemStateStorageHelper stores the given state onto the specified file system and returns a FileSerializableStateHandle.

Various fixes due to rebasing.

Changes test program to use an int accumulator which is checked at the end of the program.
This avoids to look for the "Job status changed to FINISHED." string in the stdout output,
which can sometimes not be printed if the JobExecutionResult arrives earlier than the
JobStatusChanged message at the JobClientActor.

The JobManagerRetriever sets the new leaderGatewayPortFuture directly in the notifyLeaderAddress
method instead of in one of the futures. This avoids race conditions between multiple futures
which finish in a different order than they were started. Furthermore, this replaces promises
by futures where a promise is not needed.

Add logging statement

Fix WebRuntimeMonitorITCase to use random port and proper state backend

Add ChannelHandler.Sharable to RuntimeMonitorHandler

Remove sanity check from WebInfoServer to let it work on Yarn

Squashes:
5a88d5e [tests] Add HttpTestClient for testing HTTP responses
656d6d6 Split WebMonitor and LeaderRetrievalService start up
a7e8da8 Move generated /web files to src/main/resources

Add comment to webMonitorPort attribute and make line breaks more Scalaesque

Don't block on leader retrieval and only resolve associated job manager once

Make JobManagerRetriever independent of redirecting logic

This closes #1202.

The JobClientActor is now repsonsible for receiving the JobStatus updates from
a newly elected leader. It uses the LeaderRetrievalService to be notified about
new leaders. The actor can only be used to submit a single job to the JM. Once
it received a job from the Client it tries to send it to the current leader.
If no leader is available, a connection timeout is triggered. If the job could
be sent to the JM, a submission timeout is triggered if the JobClientActor does
not receive a JobSubmitSuccess message within the timeout interval. If the
connection to the leader is lost after having submitted a job, a connection
timeout is triggered if the JobClientActor cannot reconnect to another JM within
the timeout interval. The JobClient simply awaits on the completion of the
returned future to the SubmitJobAndWait message.

Added test cases for JobClientActor exceptions

This closes #1249.

Move StateBackend enum to top level and org.apache.flink.runtime.state

Abstract blob store in blob server for recovery

This closes #1227.

Internal actor states must only be modified within the actor thread.
This avoids all the well-known issues coming with concurrency.

Fix RemoveCachedJob by introducing RemoveJob

Fix JobManagerITCase

Add removeJob which maintains the job in the SubmittedJobGraphStore

Make revokeLeadership not remove the jobs from the state backend

Fix shading problem with curator by hiding CuratorFramework in ChaosMonkeyITCase

This closes #1153.

This closes #1262

[FLINK-2842] [documentation] Remove Flink S3FileSystem, extend documentation to use Hadoop S3FileSystem.

This closes #1245

- Improve javadocs of Graph creation methods
- Add fromTuple2 creation methods
- Rename mapper parameters to vertexInitializer.
- Improve javadocs and parameter names of joinWith* methods
- Improve javadocs of neighborhood methods
- Update docs to reflect api changes

This closes #1263

This closes #1191

This closes #1217

This closes #1233.

The application status is set to failed for jobs which did not complete
successfully. If the client shuts down the YARN cluster, the
applications should be reported as SUCCESSFUL.

I'm adding this so that we have a bit more context when it fails next
time.

update test to get the directed graph as input

This closes #1250

Also reorganizing classes to use Gelly's Graph APIs.

This closes #1254

This closes #1259

This closes #1246