Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/3151#discussion_r102760735
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
---
@@ -212,6 +220,38 @@ public void start() throws Exception {
// start the job leader service
jobLeaderService.start(getAddress(), getRpcService(),
haServices, new JobLeaderListenerImpl());
+
+ // start the heartbeat manager for monitoring job manager and
resource manager
+ heartbeatManager.start(new HeartbeatListener<Void, Void>() {
+ @Override
+ public void notifyHeartbeatTimeout(final ResourceID
resourceID) {
+ runAsync(new Runnable() {
+ @Override
+ public void run() {
+ if
(jobManagerConnections.containsKey(resourceID)) {
+ log.info("Notify
heartbeat timeout with job manager {}", resourceID);
+ JobManagerConnection
jobManagerConnection = jobManagerConnections.get(resourceID);
+ if
(jobManagerConnection != null) {
+
closeJobManagerConnection(jobManagerConnection.getJobID());
+ }
+ }
+ // TODO check whether the
resource id indicates the resource manager based on resource manager connection
+ // TODO then trigger the action
of losing resource manager
--- End diff --
I assume we'll use a different `HeartbeatManager` for the resource manager
heartbeats because we have different payload there.
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
