class NodeDisconnectJobMonitorService extends AbstractLifecycleComponent implements TransportConnectionListener {

    private final TasksService tasksService;
    private final NodeLimits nodeLimits;
    private final TransportService transportService;

    private final ActionExecutor<KillJobsNodeRequest, KillResponse> killNodeAction;
    private static final Logger LOGGER = LogManager.getLogger(NodeDisconnectJobMonitorService.class);

    @Inject
    public NodeDisconnectJobMonitorService(TasksService tasksService,
                                           NodeLimits nodeLimits,
                                           TransportService transportService,
                                           Node node) {
        this(tasksService, nodeLimits, transportService, req -> node.client().execute(KillJobsNodeAction.INSTANCE, req));
    }

    @VisibleForTesting
    NodeDisconnectJobMonitorService(TasksService tasksService,
                                    NodeLimits nodeLimits,
                                    TransportService transportService,
                                    ActionExecutor<KillJobsNodeRequest, KillResponse> killNodeAction) {
        this.tasksService = tasksService;
        this.nodeLimits = nodeLimits;
        this.transportService = transportService;
        this.killNodeAction = killNodeAction;
    }

    @Override
    protected void doStart() {
        transportService.addConnectionListener(this);
    }

    @Override
    protected void doStop() {
        transportService.removeConnectionListener(this);
    }

    @Override
    protected void doClose() {
    }

    @Override
    public void onNodeConnected(DiscoveryNode node, Transport.Connection connection) {
    }

    @Override
    public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connection) {
        nodeLimits.nodeDisconnected(node.getId());
        killJobsCoordinatedBy(node);
        broadcastKillToParticipatingNodes(node);
    }

    /**
     * Broadcast the kill if *this* node is the coordinator and a participating node died
     * The information which nodes are participating is only available on the coordinator, so other nodes
     * can not kill the jobs on their own.
     *
     * <pre>
     *              n1                      n2                  n3
     *               |                      |                   |
     *           startJob 1 (n1,n2,n3)      |                   |
     *               |                      |                   |
     *               |                    *dies*                |
     *               |                                          |
     *           onNodeDisc(n2)                            onNodeDisc(n2)
     *            broadcast kill job1                   does not know which jobs involve n2
     *                  |
     *      kill job1 <-+---------------------------------->  kill job1
     *
     * </pre>
     */
    private void broadcastKillToParticipatingNodes(DiscoveryNode deadNode) {<FILL_FUNCTION_BODY>}

    /**
     * Immediately kills all jobs that were initiated by the disconnected node.
     * It is not possible to send results to the disconnected node.
     * <pre>
     *
     *              n1                      n2                  n3
     *               |                      |                   |
     *           startJob 1 (n1,n2,n3)      |                   |
     *               |                      |                   |
     *              *dies*                  |                   |
     *                                   onNodeDisc(n1)      onNodeDisc(n1)
     *                                    killJob 1            killJob1
     * </pre>
     */
    private void killJobsCoordinatedBy(DiscoveryNode deadNode) {
        List<UUID> jobsStartedByDeadNode = tasksService
            .getJobIdsByCoordinatorNode(deadNode.getId()).collect(Collectors.toList());
        if (jobsStartedByDeadNode.isEmpty()) {
            return;
        }
        tasksService.killJobs(
            jobsStartedByDeadNode,
            Role.CRATE_USER.name(),
            "Participating node=" + deadNode.getName() + " disconnected."
        );
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Killed {} jobs started by disconnected node={}", jobsStartedByDeadNode.size(), deadNode.getId());
        }
    }
}

        List<UUID> affectedJobs = tasksService
            .getJobIdsByParticipatingNodes(deadNode.getId()).collect(Collectors.toList());
        if (affectedJobs.isEmpty()) {
            return;
        }
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(
                "Broadcasting kill for {} jobs because they involved disconnected node={}",
                affectedJobs.size(),
                deadNode.getId());
        }
        List<String> excludedNodeIds = Collections.singletonList(deadNode.getId());
        KillJobsNodeRequest killRequest = new KillJobsNodeRequest(
            excludedNodeIds,
            affectedJobs,
            Role.CRATE_USER.name(),
            "Participating node=" + deadNode.getName() + " disconnected."
        );
        killNodeAction
            .execute(killRequest)
            .whenComplete(
                (resp, t) -> {
                    if (t != null) {
                        LOGGER.warn("failed to send kill request to nodes");
                    }
                }
            );
    
<methods>public void addLifecycleListener(org.elasticsearch.common.component.LifecycleListener) ,public void close() ,public org.elasticsearch.common.component.Lifecycle.State lifecycleState() ,public void removeLifecycleListener(org.elasticsearch.common.component.LifecycleListener) ,public void start() ,public void stop() <variables>protected final org.elasticsearch.common.component.Lifecycle lifecycle,private final List<org.elasticsearch.common.component.LifecycleListener> listeners