You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by beyond1920 <gi...@git.apache.org> on 2016/11/12 02:56:08 UTC
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
GitHub user beyond1920 opened a pull request:
https://github.com/apache/flink/pull/2791
[FLINK-4449] [cluster management] add Heartbeat manager in ResourceManager with TaskExecutor
This pr aims to add Heartbeat manager in ResourceManager with TaskExecutor. This pr focused on ResourceManager side, while #2770 focused on the TaskExecutor side. The main difference in this pr including:
1. Add heartbeatService which is responsible for heartbeat communication with TaskExecutor
2. Register heartbeatService to ResourceManager as a service when ResourceManagerRunner constructor a new ResourceManager instance
3. Add heartbeat RPC method in ResourceManagerGateway and TaskExecutorGateway, and implement the logic in the ResourceManager.
4. Change the TaskExecutorRegistrationSuccess because RM should tell TM its heart identifier.
5. Change some testcase in order to adapt to the modification.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/alibaba/flink jira-4449
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2791.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 #2791
----
commit 1326cc08988929263b31db8b4e8cf3d9107311d6
Author: beyond1920 <be...@126.com>
Date: 2016-10-31T09:50:15Z
ResourceManager need a heartbeat manager to monitor the connections with all registered TaskExecutor, add heartbeatService in ResourceManager to monitor liveness of taskExecutors
commit 3344e817ffa868e02778ea75ccd5c6fb40aa6670
Author: beyond1920 <be...@126.com>
Date: 2016-11-12T02:08:55Z
add heartbeat logic from resourceManger to taskManager
commit f3c432f569a642536bf03fadcbc81823d0b7ea94
Author: beyond1920 <be...@126.com>
Date: 2016-11-12T02:38:05Z
add heartbeat mechinism between rm and tm
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99581239
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/HeartbeatService.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatListener;
+import org.apache.flink.runtime.heartbeat.HeartbeatManagerSenderImpl;
+import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service which is responsible for heartbeat communication with TaskExecutor.
+ */
+public class HeartbeatService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HeartbeatService.class);
+
+ private final HeartbeatManagerSenderImpl<Void, Void> taskExecutorHeartbeatManager;
+
+ private HeartbeatListener<Void, Void> taskExecutorHeartbeatListener;
+
+ /** Resource ID of the ResourceManager */
+ private final ResourceID resourceManagerIdentify;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ public HeartbeatService(ResourceManagerConfiguration resourceManagerConfiguration, Executor executor) {
+ this.resourceManagerIdentify = ResourceID.generate();
+ this.scheduledExecutorService = Executors.newScheduledThreadPool(16);
+ this.taskExecutorHeartbeatManager = new HeartbeatManagerSenderImpl<>(resourceManagerConfiguration.getHeartbeatInterval().toMilliseconds(),
+ resourceManagerConfiguration.getTimeout().toMilliseconds(), resourceManagerIdentify, executor, scheduledExecutorService, LOG);
+ }
+
+ /**
+ * Starts the service with the given heartbeat listener
+ *
+ * @param taskExecutorHeartbeatListener heartbeat listener to listener for heartbeat actions with TaskExecutor
+ */
+ public void start(HeartbeatListener<Void, Void> taskExecutorHeartbeatListener) {
+ this.taskExecutorHeartbeatManager.start(taskExecutorHeartbeatListener);
+ }
+
+ /**
+ * Stops the service
+ *
+ * @throws Exception
+ */
+ public void stop() throws Exception {
+ try {
+ scheduledExecutorService.shutdown();
+ while (!scheduledExecutorService.awaitTermination(5000L, TimeUnit.MILLISECONDS)) {
+ // break util scheduleExecutorService terminates
+ }
+ } finally {
+ this.taskExecutorHeartbeatManager.stop();
+ }
+ }
+
+ /**
+ * Adds a TaskExecutor to monitor
+ *
+ * @param resourceID ResourceId of the taskExecutor to monitor
+ * @param taskExecutorGateway the taskExecutor to monitor
+ */
+ public void monitorTaskExecutor(ResourceID resourceID, final TaskExecutorGateway taskExecutorGateway) {
+
+ this.taskExecutorHeartbeatManager.monitorTarget(resourceID, new HeartbeatTarget<Void>() {
+
+ @Override
+ public void sendHeartbeat(ResourceID resourceID, Void payload) {
+ throw new UnsupportedOperationException("the sendHeartbeat is not supported here!");
--- End diff --
Why not?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99582774
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorRegistrationSuccess.java ---
@@ -33,16 +34,21 @@
private final InstanceID registrationId;
+ private final ResourceID resourceManagerIdentify;
+
private final long heartbeatInterval;
/**
* Create a new {@code TaskExecutorRegistrationSuccess} message.
- *
- * @param registrationId The ID that the ResourceManager assigned the registration.
- * @param heartbeatInterval The interval in which the ResourceManager will heartbeat the TaskExecutor.
+ *
+ * @param registrationId The ID that the ResourceManager assigned the registration.
+ * @param resourceManagerIdentify The identify of the ResourceManager, for heartbeat
--- End diff --
`resourceManagerId`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99582640
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerRunner.java ---
@@ -55,6 +55,7 @@ public ResourceManagerRunner(
final ResourceManagerConfiguration resourceManagerConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
final SlotManagerFactory slotManagerFactory = new DefaultSlotManager.Factory();
final JobLeaderIdService jobLeaderIdService = new JobLeaderIdService(highAvailabilityServices);
+ final HeartbeatService heartbeatService = new HeartbeatService(resourceManagerConfiguration, rpcService.getExecutor());
--- End diff --
Things which are started outside of the `ResourceManager` should also be closed outside of 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2791: [FLINK-4449] [cluster management] add Heartbeat manager i...
Posted by beyond1920 <gi...@git.apache.org>.
Github user beyond1920 commented on the issue:
https://github.com/apache/flink/pull/2791
@tillrohrmann , thanks your reply. However my partner @wangzhijiang999 had a different implementation on this jira before, he will open a new pr soon. So I would close this pr. Sorry for wasting your 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99583687
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java ---
@@ -112,22 +111,29 @@ void notifySlotAvailable(
* Unregisters an infoMessage listener
*
* @param infoMessageListenerAddress address of infoMessage listener to unregister from this resource manager
- *
*/
void unRegisterInfoMessageListener(String infoMessageListenerAddress);
/**
* shutdown cluster
+ *
* @param finalStatus
* @param optionalDiagnostics
*/
void shutDownCluster(final ApplicationStatus finalStatus, final String optionalDiagnostics);
/**
* Gets the currently registered number of TaskManagers.
- *
+ *
* @param leaderSessionId The leader session ID with which to address the ResourceManager.
* @return The future to the number of registered TaskManagers.
*/
Future<Integer> getNumberOfRegisteredTaskManagers(UUID leaderSessionId);
+
+ /*
+ * Sends the heartbeat to resource manager from task manager
+ *
+ * @param resourceID unique id of the task manager
+ */
+ void sendHeartbeatFromTaskManager(final ResourceID resourceID);
--- End diff --
clearer to name this method `heartbeatFromTaskManager` or `reportHeartbeatFromTaskManager`.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99583786
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java ---
@@ -129,4 +130,11 @@
* @return Future acknowledge if the task is successfully canceled
*/
Future<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, @RpcTimeout Time timeout);
+
+ /**
+ * request heartbeat from the resource manager
--- End diff --
It does not request a heartbeat from the resource manager but it is a heartbeat request from the resource manager.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99581786
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/HeartbeatService.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatListener;
+import org.apache.flink.runtime.heartbeat.HeartbeatManagerSenderImpl;
+import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service which is responsible for heartbeat communication with TaskExecutor.
+ */
+public class HeartbeatService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HeartbeatService.class);
+
+ private final HeartbeatManagerSenderImpl<Void, Void> taskExecutorHeartbeatManager;
+
+ private HeartbeatListener<Void, Void> taskExecutorHeartbeatListener;
+
+ /** Resource ID of the ResourceManager */
+ private final ResourceID resourceManagerIdentify;
--- End diff --
What is a `resourceManagerIdentify`? Why not calling it `resourceManagerId`?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99581018
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/HeartbeatService.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatListener;
+import org.apache.flink.runtime.heartbeat.HeartbeatManagerSenderImpl;
+import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service which is responsible for heartbeat communication with TaskExecutor.
+ */
+public class HeartbeatService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HeartbeatService.class);
+
+ private final HeartbeatManagerSenderImpl<Void, Void> taskExecutorHeartbeatManager;
+
+ private HeartbeatListener<Void, Void> taskExecutorHeartbeatListener;
+
+ /** Resource ID of the ResourceManager */
+ private final ResourceID resourceManagerIdentify;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ public HeartbeatService(ResourceManagerConfiguration resourceManagerConfiguration, Executor executor) {
+ this.resourceManagerIdentify = ResourceID.generate();
+ this.scheduledExecutorService = Executors.newScheduledThreadPool(16);
+ this.taskExecutorHeartbeatManager = new HeartbeatManagerSenderImpl<>(resourceManagerConfiguration.getHeartbeatInterval().toMilliseconds(),
+ resourceManagerConfiguration.getTimeout().toMilliseconds(), resourceManagerIdentify, executor, scheduledExecutorService, LOG);
+ }
+
+ /**
+ * Starts the service with the given heartbeat listener
+ *
+ * @param taskExecutorHeartbeatListener heartbeat listener to listener for heartbeat actions with TaskExecutor
+ */
+ public void start(HeartbeatListener<Void, Void> taskExecutorHeartbeatListener) {
+ this.taskExecutorHeartbeatManager.start(taskExecutorHeartbeatListener);
+ }
+
+ /**
+ * Stops the service
+ *
+ * @throws Exception
+ */
+ public void stop() throws Exception {
+ try {
+ scheduledExecutorService.shutdown();
+ while (!scheduledExecutorService.awaitTermination(5000L, TimeUnit.MILLISECONDS)) {
+ // break util scheduleExecutorService terminates
+ }
+ } finally {
+ this.taskExecutorHeartbeatManager.stop();
+ }
--- End diff --
I think you first have to `stop` the `taskExecutorHeartbeatManager` and then only shutdown the `scheduledExecutorService`.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99581117
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/HeartbeatService.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatListener;
+import org.apache.flink.runtime.heartbeat.HeartbeatManagerSenderImpl;
+import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service which is responsible for heartbeat communication with TaskExecutor.
+ */
+public class HeartbeatService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HeartbeatService.class);
+
+ private final HeartbeatManagerSenderImpl<Void, Void> taskExecutorHeartbeatManager;
+
+ private HeartbeatListener<Void, Void> taskExecutorHeartbeatListener;
+
+ /** Resource ID of the ResourceManager */
+ private final ResourceID resourceManagerIdentify;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ public HeartbeatService(ResourceManagerConfiguration resourceManagerConfiguration, Executor executor) {
+ this.resourceManagerIdentify = ResourceID.generate();
+ this.scheduledExecutorService = Executors.newScheduledThreadPool(16);
+ this.taskExecutorHeartbeatManager = new HeartbeatManagerSenderImpl<>(resourceManagerConfiguration.getHeartbeatInterval().toMilliseconds(),
+ resourceManagerConfiguration.getTimeout().toMilliseconds(), resourceManagerIdentify, executor, scheduledExecutorService, LOG);
+ }
+
+ /**
+ * Starts the service with the given heartbeat listener
+ *
+ * @param taskExecutorHeartbeatListener heartbeat listener to listener for heartbeat actions with TaskExecutor
+ */
+ public void start(HeartbeatListener<Void, Void> taskExecutorHeartbeatListener) {
+ this.taskExecutorHeartbeatManager.start(taskExecutorHeartbeatListener);
+ }
+
+ /**
+ * Stops the service
+ *
+ * @throws Exception
+ */
+ public void stop() throws Exception {
+ try {
+ scheduledExecutorService.shutdown();
+ while (!scheduledExecutorService.awaitTermination(5000L, TimeUnit.MILLISECONDS)) {
+ // break util scheduleExecutorService terminates
--- End diff --
What if the `scheduledExecutorService` never terminates? We should not use an potentially infinite loop.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99581410
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/HeartbeatService.java ---
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatListener;
+import org.apache.flink.runtime.heartbeat.HeartbeatManagerSenderImpl;
+import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Service which is responsible for heartbeat communication with TaskExecutor.
+ */
+public class HeartbeatService {
--- End diff --
I think this wrapper class is unnecessary. I think it would be better to directly give the `HeartbeatManager` to the `ResourceManager`. Then we save a level of indirection.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by beyond1920 <gi...@git.apache.org>.
Github user beyond1920 closed the pull request at:
https://github.com/apache/flink/pull/2791
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99582457
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java ---
@@ -93,8 +92,8 @@
* Sent by the TaskExecutor to notify the ResourceManager that a slot has become available.
*
* @param resourceManagerLeaderId The ResourceManager leader id
- * @param instanceId TaskExecutor's instance id
- * @param slotID The SlotID of the freed slot
+ * @param instanceId TaskExecutor's instance id
+ * @param slotID The SlotID of the freed slot
--- End diff --
Formatting changes
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2791: [FLINK-4449] [cluster management] add Heartbeat ma...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2791#discussion_r99582381
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java ---
@@ -831,5 +856,30 @@ public void handleError(Throwable error) {
onFatalErrorAsync(error);
}
}
+
+ private class HeartbeatListenerWithTaskExecutor implements HeartbeatListener<Void, Void> {
+
+ @Override
+ public void notifyHeartbeatTimeout(final ResourceID resourceID) {
+ log.warn("Lost heartbeat with taskExecutor {}, mark the taskExecutor as failed ", resourceID);
+ runAsync(new Runnable() {
+ @Override
+ public void run() {
+ heartbeatService.unmonitorTaskExecutor(resourceID);
--- End diff --
Why not removing the heartbeat target as part of the `notifyWorkerFailed` call?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---