public void replay(BrokerRegistrationChangeRecord record) {
        BrokerRegistration curRegistration =
            getBrokerOrThrow(record.brokerId(), record.brokerEpoch(), "change");
        BrokerRegistrationFencingChange fencingChange =
            BrokerRegistrationFencingChange.fromValue(record.fenced()).orElseThrow(
                () -> new IllegalStateException(String.format("Unable to replay %s: unknown " +
                    "value for fenced field: %d", record, record.fenced())));
        BrokerRegistrationInControlledShutdownChange inControlledShutdownChange =
            BrokerRegistrationInControlledShutdownChange.fromValue(record.inControlledShutdown()).orElseThrow(
                () -> new IllegalStateException(String.format("Unable to replay %s: unknown " +
                    "value for inControlledShutdown field: %d", record, record.inControlledShutdown())));
        Optional<List<Uuid>> directoriesChange = Optional.ofNullable(record.logDirs()).filter(list -> !list.isEmpty());
        BrokerRegistration nextRegistration = curRegistration.cloneWith(
            fencingChange.asBoolean(),
            inControlledShutdownChange.asBoolean(),
            directoriesChange
        );
        if (!curRegistration.equals(nextRegistration)) {
            changedBrokers.put(record.brokerId(), Optional.of(nextRegistration));
        }
    }
Represents changes to the cluster in the metadata image.
metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java