Add rescaling functionality to flink-connector

[tillrohrmann]

Problem description

In order to fully exploit Pravega's capability of increasing and decreasing the read parallelism, the flink-connector should be able to signal Flink whenever a change of parallelism is required. The new parallelism should depend on the new number of Pravega segments.

Suggestions for an improvement

The flink-connector needs to implement a soon to be provided RescalingPolicyFactory which can instantiate a per-operator RescalingPolicy implementation. The RescalingPolicy could have a method which is periodically polled for parallelism change requests. Whenever Flink calls into this method, the Pravega RescalingPolicy implementation could check the current number of segments to decide on a parallelism change.

This issue depends on Flink to provide the RescalingPolicy interface. In order to steer the development of this feature I would like to collect Pravega specific feedback in this thread. Once we have opened a public discussion on the Flink ML, I'll cross link to it.

[crazyzhou]

As the reactive container mode will soon come out in the Flink 1.13 release, the rescaling seems to have a new approach to do by registering more taskmanagers to the app cluster.