private void onContextsChange(FailoverContext[] contexts) {
        Set<FailedNode> oldFailedNodes = recovering.keySet();
        Set<FailedNode> newFailedNodes = Arrays.stream(contexts)
            .filter(ctx -> ctx.getTarget() == nodeId)
            .map(FailoverContext::getFailedNode)
            .collect(Collectors.toSet());

        Set<FailedNode> completedNodes = Sets.difference(oldFailedNodes, newFailedNodes);
        completedNodes.forEach(recovering::remove);

        Set<FailedNode> needFailoverNodes = Sets.difference(newFailedNodes, oldFailedNodes);
        for (FailoverContext context : contexts) {
            FailedNode failedNode = context.getFailedNode();
            if (needFailoverNodes.contains(failedNode)) {
                recovering.put(failedNode, context);
                failover(context);
            }
        }
    }
Get the failover contexts from the given metadata delta.
It returns empty if there is no update for the failover key.
core/src/main/java/kafka/automq/failover/FailoverListener.java