[GitHub] flink pull request #2141: [flink-4021] Problem of setting autoread for netty...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2141 [flink-4021] Problem of setting autoread for netty channel when more ta⦠Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration. If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html). In addition to going through the list, please provide a meaningful description of your changes. - [ ] General - The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text") - The pull request addresses only one issue - Each commit in the PR has a meaningful commit message (including the JIRA id) - [ ] Documentation - Documentation has been added for new functionality - Old documentation affected by the pull request has been updated - JavaDoc for public methods has been added - [ ] Tests & Build - Functionality added by the pull request is covered by tests - `mvn clean verify` has been executed successfully locally or a Travis build has passed â¦sks sharing the same Tcp connection You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink master Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2141.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2141 commit 7941150cc153d694b543dc92a36cc089eafcdbac Author: æ·æ± Date: 2016-06-21T09:13:37Z flink-4021:Problem of setting autoread for netty channel when more tasks sharing the same Tcp connection --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2461: [FLINK-4505][Cluster Management] Implement TaskMan...
Github user wangzhijiang999 closed the pull request at: https://github.com/apache/flink/pull/2461 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 @tillrohrmann , thank you for merging and help. If there are any following works to do related with TaskManager, you can assign to me and I am willing to do. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2770 [FLINK-4354]Implement TaskManager side of heartbeat from ResourceManager When TaskManager registers at the new ResourceManager, the SlotReport will be attached in the message. In heartbeat process, it is no need to exchange SlotReport between TaskManager and ResourceManager, so the payload is null in heartbeat message. When TaskManager' listener is notified of heartbeat timeout from ResourceManager, currently it does nothing in the event notification. And it will re-register the new ResourceManager by HA mechanism. You can merge this pull request into a Git repository by running: $ git pull https://github.com/alibaba/flink jira-4354 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2770.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2770 commit 233e63a8b92c68fc3aeeb7feae70c8e97a3e435e Author: æ·æ± Date: 2016-11-08T10:16:00Z [FLINK-4354]Implement TaskManager side of heartbeat from ResourceManager --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2798: [FLINK-4364]Implement TaskManager side of heartbea...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2798 [FLINK-4364]Implement TaskManager side of heartbeat from JobManager 1. **TaskManagerRunner** creates **HeartbeatManagerImpl** components for monitoring **JobManager** and **ResourceManager** separately when constructs the **TaskManager**. 2. **TaskManager** begins to monitor the **JobManager** when registration at new job leader successful. 3. Currently the payload will be null for interaction between **TaskManager** and **JobManager**, it can be expanded if needed later. 4. The **JobManager** side implementation will be realized in another jira. You can merge this pull request into a Git repository by running: $ git pull https://github.com/alibaba/flink jira-4364 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2798.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2798 commit ae93a8fa3f29a1d00c1864575f030f0177cf73f1 Author: æ·æ± Date: 2016-11-14T09:54:58Z [FLINK-4364]Implement TaskManager side of heartbeat from JobManager --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2906: [FLINK-5132][FLINK-5133][FLINK-5134][Core,DataStre...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2906 [FLINK-5132][FLINK-5133][FLINK-5134][Core,DataStream API,DataSet API] Fine-grained Resource Configuration Introduce the **ResourceSpec** structure to describe the different resource factors. Add new **setResource** API for both **DataStream** and **DataSet**. In the internal construction of **StreamGraph** and **JobGraph** for DataStream, set the corresponding resource and merge resource for the **JobVertex**, and the similar way in **Plan**, **OptimizedPlan** for **DataSet**. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink jira-5132 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2906.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2906 commit 73e7cca3b3540b877bf1c03d18fc77d6b69dc867 Author: æ·æ± Date: 2016-11-28T10:07:15Z [FLINK-5132][FLINK-5133][FLINK-5134][Core,DataStream API,DataSet API] Fine-grained Resource Configuration Summary: above Test Plan: NA Reviewers: yushi.wxg Subscribers: #blink Differential Revision: http://phabricator.taobao.net/D6593 commit 02b0e9d71db86921485ee27dbbb588f6c48fd19e Author: æ·æ± Date: 2016-11-30T09:13:32Z [FLINK-5132][DataSet API,DataStream API] Introduce the ResourceSpec API for DataSet and DataStream commit ceed4cb5a9798fb628e82511c08b70636acae9b0 Author: æ·æ± Date: 2016-11-30T09:43:22Z [FLINK-5132][DataSet API,DataStream API] Introduce the ResourceSpec API for DataSet and DataStream --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2907: [FLINK-5135][JM]ResourceProfile for slot request s...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2907 [FLINK-5135][JM]ResourceProfile for slot request should be expanded to correspond with ResourceSpec Currently the **ResourceProfile** only contains cpu cores and memory properties. The memory should be expanded to different types such as heap memory, direct memory and native memory which corresponds with memory in **ResourceSpec**. It contains the related un-merging codes for [FLINK-5132] [FLINK-5133] [FLINK-5134]. And the modifications for [FLINK-5135] only includes **SlotPool** and **ResourceProfile**. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink jira-5135 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2907.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2907 commit 73e7cca3b3540b877bf1c03d18fc77d6b69dc867 Author: æ·æ± Date: 2016-11-28T10:07:15Z [FLINK-5132][FLINK-5133][FLINK-5134][Core,DataStream API,DataSet API] Fine-grained Resource Configuration Summary: above Test Plan: NA Reviewers: yushi.wxg Subscribers: #blink Differential Revision: http://phabricator.taobao.net/D6593 commit 02b0e9d71db86921485ee27dbbb588f6c48fd19e Author: æ·æ± Date: 2016-11-30T09:13:32Z [FLINK-5132][DataSet API,DataStream API] Introduce the ResourceSpec API for DataSet and DataStream commit ceed4cb5a9798fb628e82511c08b70636acae9b0 Author: æ·æ± Date: 2016-11-30T09:43:22Z [FLINK-5132][DataSet API,DataStream API] Introduce the ResourceSpec API for DataSet and DataStream commit 6021b056f4de8bb2e74ff6709326b58e69547404 Author: æ·æ± Date: 2016-11-30T09:55:10Z [FLINK-5132][DataSet API,DataStream API] Introduce the ResourceSpec API for DataSet and DataStream commit 3ccd36033ff9b55c5f7ec7b4afbbefd261c85d61 Author: æ·æ± Date: 2016-11-30T10:01:16Z [FLINK-5135][JM]ResourceProfile for slot request should be expanded to correspond with ResourceSpec --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2400 [FLINK-4363] Implement TaskManager basic startup of all components in java Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration. If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html). In addition to going through the list, please provide a meaningful description of your changes. - [ ] General - The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text") - The pull request addresses only one issue - Each commit in the PR has a meaningful commit message (including the JIRA id) - [ ] Documentation - Documentation has been added for new functionality - Old documentation affected by the pull request has been updated - JavaDoc for public methods has been added - [ ] Tests & Build - Functionality added by the pull request is covered by tests - `mvn clean verify` has been executed successfully locally or a Travis build has passed You can merge this pull request into a Git repository by running: $ git pull https://github.com/alibaba/flink jira-4363 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2400.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2400 commit afc504f37ec873b3be30320c3bcbc93fcbe1aec3 Author: æ·æ± Date: 2016-08-19T09:32:10Z implement taskmanager startup basic components in java [#FLINK-4363] commit bba0f679d3b2f7b779c18b13fefd4b50edbdfb69 Author: æ·æ± Date: 2016-08-22T03:13:02Z Merge branch 'flip-6' of https://github.com/apache/flink into flink-tm_startup commit 22da4e24e155f309d2705bbb5a6d1a64744dc4e3 Author: æ·æ± Date: 2016-08-22T03:52:05Z [FLINK-4363] Implement TaskManager basic startup of all components in java --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75648894 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75649383 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) --- End diff -- It is better to
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75650354 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75651298 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75651382 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75652404 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; --- End diff -- Yes, the initial reason for this is that all the inner names still called "TaskManager", so the TaskExecutor class name can easily be be renamed to "TaskManager" to replace it in final. Otherwise we should replace all the "taskexecutor" into "taskmanager" after final replace. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2400: [FLINK-4363] Implement TaskManager basic startup of all c...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2400 Thank you for your review and advice, mxm. Another concern is whether to modify the parameters in the methods "selectNetworkInterfaceAndRunTaskManager" and "startTaskManagerComponentsAndActor" which will be invoked by "YarnTMRunner" and "LocalFlinkMiniCluster". The main differences for new TM are "RPCService" and "HighAvailabilityServices" instead of "ActorSystem" and "LeaderRetrievalService". My current implementation is only supporting these new components in constructor for easy test and retain the previous methods and parameters. Should we further modify the related parts with "ActorSystem" and "LeaderRetrievalService" next or just like this currently? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r75805009 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink issue #2400: [FLINK-4363] Implement TaskManager basic startup of all c...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2400 Hi @mxm , I pushed some modifications based on your comments, including: 1. Make var as final 2. Implement TaskExecutorConfiguration class in java instead of TaskManagerConfiguration in scala, the NetworkEnvironmentConfiguration should also be rewrote by java, but it will cause import mistake in NetworkEnvironment, so retain the current implementation. 3. Replace scala tuple2 by InetAddress, and the scala tuple4 by org.apache.flink.api.java.tuple.Tuple4 4. Remove throws in checkConfigParameter method 5. There are two public static methods "selectNetworkInterfaceAndRunTaskManager" and "startTaskManagerComponentsAndActor" that can be invoked by outside world. For startTaskManagerComponentsAndActor method, the parameters should pass "RPCService" and "HighAvailabilityServices" for constructing "TaskExecutor" directly. For "startTaskManagerComponentsAndActor" method, we will generate the default "RPCService" and "HighAvailabilityServices" or based on configuration. 6. Fix the missing ResourceID parameter description. Thank you for further suggestions! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76189334 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -94,12 +729,11 @@ public void notifyOfNewResourceManagerLeader(String newLeaderAddress, UUID newLe if (newLeaderAddress != null) { // the resource manager switched to a new leader log.info("ResourceManager leader changed from {} to {}. Registering at new leader.", - resourceManagerConnection.getResourceManagerAddress(), newLeaderAddress); --- End diff -- The RPC methods and start method have been put at the top of the class after TaskExecutor constructor. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76189450 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -36,27 +82,617 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskExecutorConfiguration taskExecutorConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskExecutorConfiguration taskExecutorConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskExecutorConfig = checkNotNull(taskExecutorConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param resourceID The id of the resource which the task manager will run on. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID) throws Exception { + + InetSocketAddress taskManagerAddress = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(taskManagerAddress.getHostName(), resourceID, taskManagerAddress.getPort(), configuration); + } + + private static InetSocketAddress selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { + String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null); + if (taskManagerHostname != null) { + LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname); +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76189882 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -36,27 +82,617 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskExecutorConfiguration taskExecutorConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskExecutorConfiguration taskExecutorConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskExecutorConfig = checkNotNull(taskExecutorConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param resourceID The id of the resource which the task manager will run on. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID) throws Exception { + + InetSocketAddress taskManagerAddress = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(taskManagerAddress.getHostName(), resourceID, taskManagerAddress.getPort(), configuration); + } + + private static InetSocketAddress selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { + String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null); + if (taskManagerHostname != null) { + LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname); +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76190061 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -36,27 +82,617 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskExecutorConfiguration taskExecutorConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskExecutorConfiguration taskExecutorConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskExecutorConfig = checkNotNull(taskExecutorConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param resourceID The id of the resource which the task manager will run on. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID) throws Exception { + + InetSocketAddress taskManagerAddress = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(taskManagerAddress.getHostName(), resourceID, taskManagerAddress.getPort(), configuration); + } + + private static InetSocketAddress selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { + String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null); + if (taskManagerHostname != null) { + LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname); +
[GitHub] flink issue #2400: [FLINK-4363] Implement TaskManager basic startup of all c...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2400 @mxm , could we remove the parameter "taskManagerHostname" from "startTaskManagerComponentsAndActor" method, I think the hostname can be got by {{TaskExecutor}} itself, no need to pass it from outsider world. To do so, it is more easier to construct {{TaskExecutor}} by factory method "startTaskManagerComponentsAndActor(configuration, resourceID,rpcService,haServices,localTaskManagerCommunication)" when design UT for testing. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76213620 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -36,27 +82,617 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskExecutorConfiguration taskExecutorConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskExecutorConfiguration taskExecutorConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskExecutorConfig = checkNotNull(taskExecutorConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param resourceID The id of the resource which the task manager will run on. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID) throws Exception { + + InetSocketAddress taskManagerAddress = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(taskManagerAddress.getHostName(), resourceID, taskManagerAddress.getPort(), configuration); + } + + private static InetSocketAddress selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { + String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null); + if (taskManagerHostname != null) { + LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname); +
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2400#discussion_r76367829 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/taskexecutor/TaskExecutor.java --- @@ -35,27 +79,634 @@ */ public class TaskExecutor extends RpcEndpoint { - /** The unique resource ID of this TaskExecutor */ + private static final Logger LOG = LoggerFactory.getLogger(TaskExecutor.class); + + /** Return code for critical errors during the runtime */ + private static final int RUNTIME_FAILURE_RETURN_CODE = 2; + + /** The name of the TaskManager actor */ + private static final String TASK_MANAGER_NAME = "taskmanager"; + + /** The unique resource ID of this TaskManager */ private final ResourceID resourceID; /** The access to the leader election and metadata storage services */ private final HighAvailabilityServices haServices; - // - resource manager + /** The task manager configuration */ + private final TaskManagerConfiguration taskManagerConfig; + + /** The connection information of the task manager */ + private final InstanceConnectionInfo connectionInfo; + + /** The I/O manager component in the task manager */ + private final IOManager ioManager; + + /** The memory manager component in the task manager */ + private final MemoryManager memoryManager; + + /** The network component in the task manager */ + private final NetworkEnvironment networkEnvironment; + + /** The number of slots in the task manager, should be 1 for YARN */ + private final int numberOfSlots; + // - resource manager private TaskExecutorToResourceManagerConnection resourceManagerConnection; // public TaskExecutor( + TaskManagerConfiguration taskManagerConfig, + ResourceID resourceID, + InstanceConnectionInfo connectionInfo, + MemoryManager memoryManager, + IOManager ioManager, + NetworkEnvironment networkEnvironment, + int numberOfSlots, RpcService rpcService, - HighAvailabilityServices haServices, - ResourceID resourceID) { + HighAvailabilityServices haServices) { super(rpcService); - this.haServices = checkNotNull(haServices); + this.taskManagerConfig = checkNotNull(taskManagerConfig); this.resourceID = checkNotNull(resourceID); + this.connectionInfo = checkNotNull(connectionInfo); + this.memoryManager = checkNotNull(memoryManager); + this.ioManager = checkNotNull(ioManager); + this.networkEnvironment = checkNotNull(networkEnvironment); + this.numberOfSlots = checkNotNull(numberOfSlots); + this.haServices = checkNotNull(haServices); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param taskManagerClass The actor class to instantiate. +* Allows to use TaskManager subclasses for example for YARN. +*/ + public static void selectNetworkInterfaceAndRunTaskManager( + Configuration configuration, + ResourceID resourceID, + Class taskManagerClass) throws Exception { + + Tuple2 tuple2 = selectNetworkInterfaceAndPort(configuration); + + runTaskManager(tuple2._1(), resourceID, tuple2._2(), configuration, taskManagerClass); + } + + private static Tuple2 selectNetworkInterfaceAndPort(Configuration configuration) + throws Exception { +
[GitHub] flink issue #2400: [FLINK-4363] Implement TaskManager basic startup of all c...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2400 @mxm, I already push my changes and solve the conflicts based on the new flip-6 branch. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2400: [FLINK-4363] Implement TaskManager basic startup of all c...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2400 @mxm , I have created the FLINK-4505 for TaskExecutorFactory issue. As you mentioned, it should be an abstract class and provide an abstract method maybe called 'createAndStartTaskExecutor()'. There may be at least three different specific factories(testing, yarn, standalone) extend TaskExecutorFactory to implement the method 'createAndStartTaskExecutor'. The parameters in constructor of specific factories are different based on different modes. For example: for StandaloneTaskExecutorFactory, the constructor parameter should be (Configuration configuration,ResourceID resourceID, RpcService rpcService,String taskManagerHostname,HighAvailabilityServices haServices,boolean localTaskManagerCommunication), and in the 'createAndStartTaskExecutor()' method it can invoke âstartTaskManagerComponentsAndActor' method in TaskExecutor to bring up TaskExecutor. Do you have any other advices, then I can start this subtask later. https://issues.apache.org/jira/browse/FLINK-4505 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2400: [FLINK-4363] Implement TaskManager basic startup o...
Github user wangzhijiang999 closed the pull request at: https://github.com/apache/flink/pull/2400 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2461: [FLINK-4505][Cluster Management] Implement TaskMan...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/2461 [FLINK-4505][Cluster Management] Implement TaskManagerFactory to brin⦠Implement TaskExecutorFactory that should be an abstract class with the helper methods to bring up the TaskManager. The factory can be implemented by some classes to start a TaskManager in different modes (testing, standalone, yarn). You can merge this pull request into a Git repository by running: $ git pull https://github.com/alibaba/flink jira-4505 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/2461.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #2461 commit 21c4b408bb0138b41e49a439f6d07f9a0ca9 Author: æ·æ± Date: 2016-09-02T10:00:49Z [FLINK-4505][Cluster Management] Implement TaskManagerFactory to bring up TaskManager for different modes Summary: above Test Plan: NA Reviewers: kete.yangkt Subscribers: #blink Differential Revision: http://phabricator.taobao.net/D5606 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 @mxm , The {{StandaloneTaskExecutorFactory}} can be used for mini cluster or testing mode I think, and the {{YarnTaskExecutorFactory}} used for yarn mode. After you confirm the implementation, I can add some testings for the factory if needed. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2461: [FLINK-4505][Cluster Management] Implement TaskMan...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2461#discussion_r77329068 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/YarnTaskExecutorFactory.java --- @@ -0,0 +1,198 @@ +package org.apache.flink.runtime.taskexecutor; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.configuration.ConfigConstants; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.IllegalConfigurationException; +import org.apache.flink.runtime.akka.AkkaUtils; +import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.highavailability.HighAvailabilityServices; +import org.apache.flink.runtime.leaderelection.LeaderElectionService; +import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService; +import org.apache.flink.runtime.rpc.akka.AkkaRpcService; +import org.apache.flink.runtime.taskmanager.MemoryLogger; +import org.apache.flink.runtime.util.LeaderRetrievalUtils; +import org.apache.flink.util.NetUtils; + +import akka.actor.ActorSystem; +import akka.util.Timeout; + +import scala.Some; +import scala.Tuple2; +import scala.concurrent.duration.FiniteDuration; + +import com.typesafe.config.Config; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; + +/** + * An factory for creating {@link TaskExecutor} and starting it in yarn mode. + */ +public class YarnTaskExecutorFactory extends TaskExecutorFactory { + + public YarnTaskExecutorFactory(Configuration configuration, ResourceID resourceID) { + super(configuration, resourceID); + } + + @Override + public TaskExecutor createAndStartTaskExecutor() throws Exception { + return selectNetworkInterfaceAndRunTaskManager(configuration, resourceID); + } + + /** +* Starts and runs the TaskManager. +* +* This method first tries to select the network interface to use for the TaskManager +* communication. The network interface is used both for the actor communication +* (coordination) as well as for the data exchange between task managers. Unless +* the hostname/interface is explicitly configured in the configuration, this +* method will try out various interfaces and methods to connect to the JobManager +* and select the one where the connection attempt is successful. +* +* After selecting the network interface, this method brings up an actor system +* for the TaskManager and its actors, starts the TaskManager's services +* (library cache, shuffle network stack, ...), and starts the TaskManager itself. +* +* @param configurationThe configuration for the TaskManager. +* @param resourceID The id of the resource which the task manager will run on. +*/ + private TaskExecutor selectNetworkInterfaceAndRunTaskManager( --- End diff -- The parameters provided from different modes are different as I referred to previous implementation for **TaskManager**. For example, **YarnTaskManagerRunner** invokes the method "selectNetworkInterfaceAndRunTaskManager", and **ForkableFlinkMiniCluster** invokes the method "startTaskManagerComponentsAndActor" to start **TaskManager** before. So I retained the previous ways for different modes. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 @tillrohrmann , the current implementation is just follow the previous process for yarn and standalone modes. And I agree your opinion and actually it is not very clear for current ways. I think first we should re-define and confirm the parameters passed to the "createTaskManager" method for different factories, then I can fix the inner processes based on the input parameters. Would you suggest the specific parameters passed for different factories? Thank you! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 @tillrohrmann , I tried to understand your idea as follows: 1. Provide specific TaskExecutorFactory class instead of abstract factory for both standalone/yarn mode. 2. Network selection and RPC service creation methods should be pulled out from factory, maybe remove to some utility class?(for JM reuse) 3. The constructor of TaskExecutorFactory supports many different parameters: such as TaskExecutorFactory(Configuration, ResourceID) TaskExecutorFactory(RpcService, HighAvailabilityServices) TaskExecutorFactory(hostname, port) TaskExecutorFactory(Configuration, ResourceID, RpcService, HighAvailabilityServices,hostname, port), The above three constructors for partial parameters can be transformed into the fourth final constructor , all the missing parameters can be generated by internal default value. 4. The TaskExecutorFactory supports the method "createTaskManger" to bring up TaskManger for outside world, and this method will construct the related components(TaskManagerConfiguration, NetworkManager, IOManager, MemoryManager) 5. For testing mode, construct the TestingTaskExecutorFactory to pass all the components explicitly, including (ResourceID,MemoryManager,IOManager,NetworkEnvironment,numberOfSlots, RpcService,HighAvailabilityServices), the TaskManagerConfiguration should be passed from outside or generate implicitly? 6. In addition, the localTaskManagerCommunication parameter is needed before to decide whether to create NettyConfig for standalone or yarn mode. Now I will remove this parameter to create ~~NettyConfig~~ always. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 Yes, that also make sense. For testing purpose it is very clear, so we do not need do anything currently, all the components in `TaskManager` constructor can be mocked implicitly. For `TaskManagerRunner`, the purpose is to pull out the initialization of related components from `TaskManager` to make it logic clear. Just one issue to be confirmed, we should provide more static methods of different parameter units for outside world or just one static method such as 'selectNetworkInterfaceAndRunTaskManager(`Configuration` configuration,`ResourceID` resourceID) '? I think providing more methods with different parameters may be reasonable, because some parameters such as 'hostname','port', `RpcService`, `HighAvailabilityServices` may want to be passed from outside. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2461: [FLINK-4505][Cluster Management] Implement TaskManagerFac...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2461 @tillrohrmann , I created the `TaskExecutorRunner` for constructing the related components for starting `TaskExecutor`, and removed the factory class. Currently only the `ResourceID` and 'Configuration` parameters must be passed to the method `startComponents`, other parameters are optional. After you confirm the way , I will write some testings based on it. The `MemoryLogger` is removed from `TaskExecutorRunner` temporarily. The `ActorGateway` should not be passed to the constructor of `MemoryLogger` directly I think. If you agree, I will re-structure the `MemoryLogger` in another jira. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3113: [FLINK-4912] Introduce RECONCILIATING state in Exe...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3113 [FLINK-4912] Introduce RECONCILIATING state in ExecutionGraph and Exe⦠This is part of the non-disruptive JobManager failure recovery. Add a JobStatus and ExecutionState {{RECONCILING}}. If a job is started on a JobManager for master recovery, the job status with all the executions transition to {{RECONCILING}} state. From {{RECONCILING}}, execution can go to any existing task states (execution reconciled with TaskManager). You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-4912 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3113.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3113 commit 0fbd628b9b8817fd1b71faca92d87c56213d79f6 Author: æ·æ± Date: 2017-01-13T08:41:37Z [FLINK-4912] Introduce RECONCILIATING state in ExecutionGraph and Execution for JobManager failure recovery --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2907: [FLINK-5135][JM]ResourceProfile for slot request s...
Github user wangzhijiang999 closed the pull request at: https://github.com/apache/flink/pull/2907 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2906: [FLINK-5132][FLINK-5133][FLINK-5134][Core,DataStre...
Github user wangzhijiang999 closed the pull request at: https://github.com/apache/flink/pull/2906 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3114: [FLINK-5132]Introduce the ResourceSpec for groupin...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3114 [FLINK-5132]Introduce the ResourceSpec for grouping different resourc⦠This is part of the fine-grained resource configuration. The current resource factors include cpu cores, heap memory, direct memory, native memory and state size. The **ResourceSpec** will provide some basic constructions for grouping different resource factors as needed and the construction can also be expanded easily for further requirements. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-5132 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3114.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3114 commit 8689e0d4f81de9f2b704d425aa56ca3f5e792f29 Author: æ·æ± Date: 2017-01-13T10:40:47Z [FLINK-5132]Introduce the ResourceSpec for grouping different resource factors in API --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3113: [FLINK-4912] Introduce RECONCILIATING state in Exe...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3113#discussion_r95975088 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/execution/ExecutionState.java --- @@ -25,16 +25,23 @@ * {@code * * CREATED -> SCHEDULED -> DEPLOYING -> RUNNING -> FINISHED - * || | - * || +--+ - * |V V - * | CANCELLING -+> CANCELED - * | | - * +-+ + *| || | + *| || +--+ + *| |V V + *| | CANCELLING -+> CANCELED + *| | | + *|+-+ + *| + *| ... -> FAILED + * V + *RECONCILING -> RUNNING | FINISHED | CANCELED | FAILED * - * ... -> FAILED * } * + * It is possible to enter the {@code RECONCILING} state from {@code CREATED} --- End diff -- Thank you for suggestions of the format. :) --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3125: [FLINK-5499][JobManager]Reuse the resource locatio...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3125 [FLINK-5499][JobManager]Reuse the resource location of prior executio⦠Currently when schedule execution to request to allocate slot from **SlotPool**, the **TaskManagerLocation** parameter is empty collection. So for task fail over scenario, the new execution attempt may be deployed to different task managers. If setting rockDB as state backend, the performance is better if the data can be restored from local machine. So we try to reuse the **TaskManagerLocation** of prior execution attempt when allocating slot from **SlotPool**. If the **TaskManagerLocation** is empty from prior executions, the behavior is the same with current status. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-5499 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3125.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3125 commit ab2e24ae7e82be45359f249670f72664226ec18c Author: æ·æ± Date: 2017-01-16T09:28:19Z [FLINK-5499][JobManager]Reuse the resource location of prior execution attempt in allocating slot --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3151 [FLINK-4364][runtime]mplement TaskManager side of heartbeat from JobM⦠1. **TaskManagerRunner** creates **HeartbeatManagerImpl** components when constructs the **TaskManager**. 2. **TaskManager** begins to monitor the JobManager when registration at new job leader successful. 3. Currently the payload will be null for interaction between **TaskManager** and **JobManager**, it can be expanded if needed later. Also the logic of heartbeat timeout is not implemented yet, and maybe realized in failover detection issue. 4. The **JobManager** side implementation will be realized in another jira. @tillrohrmann for review please! You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-4364 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3151.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3151 commit 362f717b0ec50d8cc864d2f2efd432a5efbfd24a Author: æ·æ± Date: 2017-01-18T11:08:19Z [FLINK-4364][runtime]mplement TaskManager side of heartbeat from JobManager --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2798: [FLINK-4364]Implement TaskManager side of heartbea...
Github user wangzhijiang999 closed the pull request at: https://github.com/apache/flink/pull/2798 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3113: [FLINK-4912] Introduce RECONCILIATING state in ExecutionG...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3113 @StephanEwen , thank you for the concrete suggestions. Sorry for delay response because of Chinese Spring Festival Holiday. I have considered and added some tests to validate the state transitions of the state machine related with the later processes which would be submitted in the following PRs together. I totally agree with the consideration of the above possible state transitions. And I plan to give a detail explanation of my implementation in another jira soon. It is actually a bit complex to do that ,so I try to break them down into small ones in order to review and merge quickly. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3125: [FLINK-5499][JobManager]Reuse the resource location of pr...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3125 @StephanEwen Yes, the current concern is only focusing on state restore performance. This PR does not consider all the scenarios and it may be only the first step for the slot location implementation. If the location do not exist, it can add other strategies to decide the locations, such as co-loated by input for batch job as you mentioned. And it can be the second step for the implementation. Wish your further comments! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3113: [FLINK-4912] Introduce RECONCILIATING state in ExecutionG...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3113 @StephanEwen , I already created #5703 for further detail recovery process and it may cover your considerations. Wish your further response, thank you! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99731288 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java --- @@ -218,4 +218,12 @@ void failSlot(final ResourceID taskManagerId, final TaskManagerLocation taskManagerLocation, final UUID leaderId, @RpcTimeout final Time timeout); + + /** +* Send the heartbeat to job manager from task manager +* +* @param resourceID unique id of the task manager +* @param payload the payload information from the task manager +*/ + void heartbeatFromTaskManager(final ResourceID resourceID, final Object payload); --- End diff -- Currently we have not considered which specific payload information should be attached with the heartbeat, so use the object to work around. It should be changed to specific type if confirmation. Do you think which payload should be attached necessary currently? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99732047 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -1058,6 +1092,30 @@ public void handleError(Throwable throwable) { } } + /** +* The heartbeat listener for JobManager and ResourceManager, they can be distinguished by ResourceID +* and trigger different processes. +*/ + private final class JMRMHeartbeatListener implements HeartbeatListener { --- End diff -- From TaskExecutor side, it will monitor the JM and RM that initiate the heartbeat, so JMRMHeartbeatListener indicates the heartbeat timeout with JM or RM, the current PR just shows the interaction with JM part. I am supposed to submit the RM part in another PR, maybe submit both sides together will help to understand easily. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99732610 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -1058,6 +1092,30 @@ public void handleError(Throwable throwable) { } } + /** +* The heartbeat listener for JobManager and ResourceManager, they can be distinguished by ResourceID +* and trigger different processes. +*/ + private final class JMRMHeartbeatListener implements HeartbeatListener { + + JMRMHeartbeatListener() { + } + + @Override + public void notifyHeartbeatTimeout(final ResourceID resourceID) { + log.info("Notify heartbeat timeout for resourceID {}", resourceID); --- End diff -- Yes, it actually should trigger some actions with timeout. Currently I did not submit this part because I think it is related with failure detection logic and supposed to submit in another PR. To make the heartbeat mechanism complete, I will add this part in the following modifications. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99733319 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java --- @@ -112,6 +114,13 @@ public TaskManagerRunner( // Initialize the TM metrics TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, taskManagerServices.getNetworkEnvironment()); + HeartbeatManagerImpl heartbeatManager = new HeartbeatManagerImpl( + taskManagerConfiguration.getTimeout().toMilliseconds(), + resourceID, + executor, + Executors.newSingleThreadScheduledExecutor(), --- End diff -- I agree with that, and initially I want to reuse the executor in **RPCService** but it needs some modifications. I will introduce it in Runner in order to share among components. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99734627 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -741,6 +763,18 @@ private void establishJobManagerConnection(JobID jobId, JobMasterGateway jobMast jobManagerTable.put(jobId, associateWithJobManager(jobMasterGateway, jobManagerLeaderId, registrationSuccess.getBlobPort())); } + heartbeatManager.monitorTarget(registrationSuccess.getResourceID(), new HeartbeatTarget() { + @Override + public void sendHeartbeat(ResourceID resourceID, Object payload) { + jobMasterGateway.heartbeatFromTaskManager(resourceID, payload); + } + + @Override + public void requestHeartbeat(ResourceID resourceID, Object payload) { + throw new UnsupportedOperationException("Should never call requestHeartbeat in task manager."); --- End diff -- My previous understanding is that the heartbeat is only requested from RM and JM to TM, and the TM will only response the heartbeat. Do you mean the heartbeat can both request from both sides? If to do so, the TM also needs to schedule a heartbeat request at interval time. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3151: [FLINK-4364][runtime]mplement TaskManager side of heartbe...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3151 @tillrohrmann . Thank you for detail review and comments! This PR just submit the heartbeat logic in TM side, because there is already a jira of JM heartbeat side. For my implementation, the JM initiates the heartbeat with **HeartbeatManagerSenderImpl** and the TM responses the heartbeat with **HeartbeatManagerImpl**. So the heartbeat process is one-way. I think it is better to submit the JM heartbeat logic in this PR in order to understand easily. I will modify this PR soon, and for testing there already exists the UT for basic heartbeat logic. Do you mean to add some ITCases? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r99736386 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -126,6 +129,9 @@ /** The metric registry in the task manager */ private final MetricRegistry metricRegistry; + /** The heartbeat manager for job manager and resource manager in the task manager */ + private final HeartbeatManagerImpl heartbeatManager; --- End diff -- Yes, the current **HeartbeatManagerImpl** is enough to use, so I did not wrapper it. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2770#discussion_r99736682 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java --- @@ -340,7 +344,8 @@ public RegistrationResponse apply(RegistrationResponse registrationResponse, Thr final UUID resourceManagerLeaderId, final String taskExecutorAddress, final ResourceID resourceID, - final SlotReport slotReport) { + final SlotReport slotReport) + { --- End diff -- It may be a misoperation, I will be careful next time. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2770#discussion_r99737093 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java --- @@ -137,6 +140,7 @@ public ResourceManager( this.jobManagerRegistrations = new HashMap<>(4); this.taskExecutors = new HashMap<>(8); this.leaderSessionId = null; + this.resourceID = ResourceID.generate(); --- End diff -- Yes, in the current implementation it is determined by **ResourceManagerRunner**.This PR is submitted long time ago and some dependent work is not complete at that time. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2770#discussion_r99737151 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java --- @@ -122,4 +122,11 @@ void notifySlotAvailable( * @param optionalDiagnostics */ void shutDownCluster(final ApplicationStatus finalStatus, final String optionalDiagnostics); + + /** +* sends the heartbeat to resource manager from task manager +* @param resourceID unique id of the task manager +* @param payload the payload information of the task manager +*/ + void sendHeartbeatFromTaskManager(final ResourceID resourceID, final Object payload); --- End diff -- agree with it --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2770#discussion_r99737204 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -1099,4 +1121,23 @@ public void run() { }); } } + + private final class ResourceManagerHeartbeatListener implements HeartbeatListener { + + ResourceManagerHeartbeatListener() { + } + + @Override + public void notifyHeartbeatTimeout(ResourceID resourceID) { + } --- End diff -- It will be added in the new modifications --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #2770: [FLINK-4354]Implement TaskManager side of heartbea...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/2770#discussion_r99737255 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java --- @@ -99,6 +101,13 @@ public TaskManagerRunner( // Initialize the TM metrics TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, taskManagerServices.getNetworkEnvironment()); + HeartbeatManagerImpl heartbeatManager = new HeartbeatManagerImpl( + taskManagerConfiguration.getTimeout().toMilliseconds(), + resourceID, + executor, + Executors.newSingleThreadScheduledExecutor(), --- End diff -- It is already added in new modifications --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #2770: [FLINK-4354]Implement TaskManager side of heartbeat from ...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/2770 @tillrohrmann , thank you for review! Actually this PR is out-of-date based on flip-6, and my following modifications do not update this PR. This PR should be re-submitted based on master and I forgot to close it. Some above suggestions have already been fixed in the current implementation, and I want to merge this part in #3151 in order to be reviewed easily. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r100030725 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java --- @@ -741,6 +763,18 @@ private void establishJobManagerConnection(JobID jobId, JobMasterGateway jobMast jobManagerTable.put(jobId, associateWithJobManager(jobMasterGateway, jobManagerLeaderId, registrationSuccess.getBlobPort())); } + heartbeatManager.monitorTarget(registrationSuccess.getResourceID(), new HeartbeatTarget() { + @Override + public void sendHeartbeat(ResourceID resourceID, Object payload) { + jobMasterGateway.heartbeatFromTaskManager(resourceID, payload); + } + + @Override + public void requestHeartbeat(ResourceID resourceID, Object payload) { + throw new UnsupportedOperationException("Should never call requestHeartbeat in task manager."); --- End diff -- If I understand correctly the TM can request the heartbeat to itself and the JM also can send the heartbeat to itself? If it is true, the **requestHeartbeat** will never be called from current logic in TM side. Or I can just remove the **UnsupportedOperationException** to not limitation and leave this method empty currently? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r100031031 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java --- @@ -112,6 +114,13 @@ public TaskManagerRunner( // Initialize the TM metrics TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, taskManagerServices.getNetworkEnvironment()); + HeartbeatManagerImpl heartbeatManager = new HeartbeatManagerImpl( + taskManagerConfiguration.getTimeout().toMilliseconds(), + resourceID, + executor, + Executors.newSingleThreadScheduledExecutor(), --- End diff -- Do you have further suggestions where to introduce the **ScheduledExecutorService** seems suitable? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3151: [FLINK-4364][runtime]mplement TaskManager side of heartbe...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3151 @tillrohrmann , I am preparing the testing code and can submit the updates this week. Thank you for continuous help! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3151: [FLINK-4364][runtime]mplement TaskManager side of ...
Github user wangzhijiang999 commented on a diff in the pull request: https://github.com/apache/flink/pull/3151#discussion_r100222823 --- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java --- @@ -112,6 +114,13 @@ public TaskManagerRunner( // Initialize the TM metrics TaskExecutorMetricsInitializer.instantiateStatusMetrics(taskManagerMetricGroup, taskManagerServices.getNetworkEnvironment()); + HeartbeatManagerImpl heartbeatManager = new HeartbeatManagerImpl( + taskManagerConfiguration.getTimeout().toMilliseconds(), + resourceID, + executor, + Executors.newSingleThreadScheduledExecutor(), --- End diff -- Yes, I totally agree with that and the advantages can avoid bringing extra thread pool and run all the PRC messages in the uniform executor if **RPCService** can handle well. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3151: [FLINK-4364][runtime]mplement TaskManager side of heartbe...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3151 @tillrohrmann , I have submitted the updates that may cover your suggestions. There are still two issues that I am not confirmed. First, for heartbeat interval and timeout default values in **ConfigConstants**, they are not invalid currently and you can modify it with your professional experience. Second, the introduction of **ScheduledExecutorService** in **RPCService**, my initial idea is trying to use the existing scheduler in **RPCService**, but it can not be got from **AkkaRPCService** implementation. Another way is to replace the current **ScheduledExecutorService** parameter with **RPCService** in construction of **HeartbeatManagerSenderImpl**, and the **RPCService** can also schedule the heartbeat request. But the return value of **scheduleRunnable** method in **RPCService** is conflict with that in **HeartbeatManagerSenderImpl**. So I just bring another single thread pool in **RPCService** for use currently. Maybe the number of threads in pool can refer to number of cpu cores. Maybe there are still something to be polished, and I am willing for further modifications by your comments. BTW, the heartbeat interaction between TM and RM will be submitted in another PR after this confirmation because of some common points. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3303: [FLINK-5133][core] Add new setResource API for Dat...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3303 [FLINK-5133][core] Add new setResource API for DataStream and DataSet This is part of the fine-grained resource configuration. For **DataStream**, the **setResource** API will be setted onto **SingleOutputStreamOperator** similar with other existing properties like parallelism, name, etc. For **DataSet**, the **setResource** API will be setted onto **Operator** in the similar way. There are two parameters described with minimum **ResourceSpec** and maximum **ResourceSpec** separately in the API for considering dynamic resource resize in future improvements. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-5133 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3303.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3303 commit de7c4c75c9f1c80bf7e529a1ff05fa870c2df1cb Author: æ·æ± Date: 2017-02-14T04:37:18Z [FLINK-5133][core] Add new setResource API for DataStream and DataSet --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3303: [FLINK-5133][core] Add new setResource API for DataStream...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3303 @StephanEwen , this PR includes the new API that would be visible to user, but it can not work completely because the following codes in runtime have not been submitted. In order not to confuse users, this PR would be fixed to hide the API temporarily before merging into master. What do you think? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3340: [FLINK-5703][runtime]Job manager failure recovery ...
GitHub user wangzhijiang999 opened a pull request: https://github.com/apache/flink/pull/3340 [FLINK-5703][runtime]Job manager failure recovery via reconciliation with TaskManager reports This is part of [Non-disruptive JobManager Failures via Reconciliation ](https://issues.apache.org/jira/browse/FLINK-4911). The design doc for this part is attached in the JIRA and it mainly contains the following work: - **RECONCILING** state transition for job status and execution. - The data structure recovery for **ExecutionGraph**, **ExecutionVertex**,**Execution** based on **TaskManager** reports. Two parts are left for the whole feature: - The related modifications in **TaskManger** side will be submitted in another JIRA, including not cancel the tasks when notified job leader changed and report the task status to **JobManager**, etc. - This PR will not make an effect on the current master branch, so it is safe to merge. The reconcile logic should be triggered by **JobManagerRunner** when grants leadership, but it is dependent on [Determine whether the job starts from last JobManager failure](https://issues.apache.org/jira/browse/FLINK-5501) , so I will modify the **JobManagerRunner** logic after [FLINK-5501](https://issues.apache.org/jira/browse/FLINK-5501) is merged. You can merge this pull request into a Git repository by running: $ git pull https://github.com/wangzhijiang999/flink FLINK-5703 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3340.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3340 commit 7337754bd8d12808a328b55e184b0fe59cf6e11d Author: æ·æ± Date: 2017-02-17T08:41:44Z [FLINK-5703][runtime]Job manager failure recovery based on reconciliation with TaskManager reports --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink issue #3340: [FLINK-5703][runtime]Job manager failure recovery via rec...
Github user wangzhijiang999 commented on the issue: https://github.com/apache/flink/pull/3340 @StephanEwen , wish your reviews. Thank you for any comments! --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---