@SuppressWarnings("checkstyle:npathcomplexity")
    public ClusterImage apply() {
        Map<Integer, BrokerRegistration> newBrokers = new HashMap<>(image.brokers().size());
        for (Entry<Integer, BrokerRegistration> entry : image.brokers().entrySet()) {
            int nodeId = entry.getKey();
            Optional<BrokerRegistration> change = changedBrokers.get(nodeId);
            if (change == null) {
                newBrokers.put(nodeId, entry.getValue());
            } else if (change.isPresent()) {
                newBrokers.put(nodeId, change.get());
            }
        }
        for (Entry<Integer, Optional<BrokerRegistration>> entry : changedBrokers.entrySet()) {
            int nodeId = entry.getKey();
            Optional<BrokerRegistration> brokerRegistration = entry.getValue();
            if (!newBrokers.containsKey(nodeId)) {
                if (brokerRegistration.isPresent()) {
                    newBrokers.put(nodeId, brokerRegistration.get());
                }
            }
        }
        Map<Integer, ControllerRegistration> newControllers = new HashMap<>(image.controllers().size());
        for (Entry<Integer, ControllerRegistration> entry : image.controllers().entrySet()) {
            int nodeId = entry.getKey();
            Optional<ControllerRegistration> change = changedControllers.get(nodeId);
            if (change == null) {
                newControllers.put(nodeId, entry.getValue());
            } else if (change.isPresent()) {
                newControllers.put(nodeId, change.get());
            }
        }
        for (Entry<Integer, Optional<ControllerRegistration>> entry : changedControllers.entrySet()) {
            int nodeId = entry.getKey();
            Optional<ControllerRegistration> controllerRegistration = entry.getValue();
            if (!newControllers.containsKey(nodeId)) {
                if (controllerRegistration.isPresent()) {
                    newControllers.put(nodeId, controllerRegistration.get());
                }
            }
        }
        Integer nextNodeId = this.nextNodeId >= 0 ? this.nextNodeId : image.nextNodeId();
        return new ClusterImage(newBrokers, newControllers, nextNodeId);
    }
Represents changes to the cluster in the metadata image.
metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java