This closes #2166

Guard test for ChainedAllReduceDriver

This closes #2156.

This closes #2160

Introduces a custom TimeServiceProvider to the StreamTask.
This is responsible for defining and updating the current
processingtime for a task and handling all related action,
such as registering timers for actions to be executed in
the future.

Before, a job would fail if object reuse was enabled and multiple
operators were chained to one upstream operator. Now, we create shallow
copies in BroadcastingOutputCollector and DirectedOutput if object reuse
is enabled.

This closes #2112

This closes #2134

This closes #2145

This closes #2105

This closes #2163

This closes #2165

This closes #2079

- enables actual checking of output which didn't work until now
- uses regexp to check for prohibited strings

This closes #2125

These timeouts can be recovered and don't have to be logged as error
with an Exception.

This closes #2137

This closes #2138

Jar arguments with a single '-' were not parsed correctly if options were
present. For example, in `./flink run <options> file.jar -arg value` the
jar arguments would be parsed as "arg" and "value". Interestingly, this only
happened when <options> were present.

The issue has been fixed in commons-cli 1.3.1.

A test case was added to test for regressions.

This closes #2139

This removes repeated printing of messages retrieved from the Yarn
cluster. Only new messages are printed.

- reduce waiting time between subsequent cluster queries

This closes #2147

- remove ActorRunner thread, print status in finalizeCluster instead
- prevent premature shutdown of actor system in shutdown method
- prevent timeout exceptions due to poisoning the ApplicationClient

The SUSPENDED job status is a new ExecutionGraph state which can be reached from all
non-terminal states when calling suspend on the ExecutionGraph. Unlike the FAILED,
FINISHED and CANCELED state, the SUSPENDED state does not trigger the deletion of the
job from the HA storage. Therefore, this state can be used to handle the loss of
leadership or the shutdown of a JobManager so that the ExecutionGraph is stopped but
can still be recovered. SUSPENDED is also a terminal state but it can be differentiated as
a locally terminal state from FAILED, CANCELED and FINISHED which are globally
terminal states.

Add test case for suspend signal

Add test case for suspending restarting job

Add test case for HA job recovery when losing leadership

Add online documentation for the job status

Add ASF license header to job_status.svg

Not throw exception when calling ExecutionGraph.restart and job is in state SUSPENDED

This closes #2096.

A job can get stuck in FAILING if fail is called on a restarting job which has
not yet reset its ExecutionJobVertices, because these vertices would not call
jobVertexInFinalState. This method, however, must be called in order to transition
from FAILING to FAILED.

Accept state FAILED when calling ExecutionGraph.restart

This closes #2095.

This closes #2088.

- Adds two new config keys for the start up scripts:
    * env.java.opts.jobmanager
    * env.java.opts.taskmanager
- These are used in addition to the regular env.java.opts for the JobManager
  and TaskManager, respectively.
- Current behaviour is not changed. This does not address the JMX port
  setting, because that is handled differently.

This closes #2143.

This closes #2069.

- Adds a deserialization schema from byte[] to Row to be used in conjunction
  with the Table API.

This closes #1996.

Update ValueTypeInfo to use specialized serializers and comparators,
many of which were already present.

This closes #1983

This closes #2142

[FLINK-4097] Fix NullPointerException on CassandraSinkBase and CassandraTupleWriteAheadSink's close()

This closes #2144

This closes #2111

This closes #2136

Create a task_id message in PythonStreamer that is passed through
to the underlying process and included in the RuntimeContext, where
it is accessible to the user and to functions.

This closes #2130

This closes #2133

By default this is set to disabled because there are known issues when
users specify a custom TypeInformation.