ControllerResult<BrokerHeartbeatReply> processBrokerHeartbeat(
        BrokerHeartbeatRequestData request,
        long registerBrokerRecordOffset
    ) {
        int brokerId = request.brokerId();
        long brokerEpoch = request.brokerEpoch();
        clusterControl.checkBrokerEpoch(brokerId, brokerEpoch);
        BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
        BrokerControlStates states = heartbeatManager.calculateNextBrokerState(brokerId,
            request, registerBrokerRecordOffset, () -> brokersToIsrs.hasLeaderships(brokerId));
        List<ApiMessageAndVersion> records = new ArrayList<>();
        if (states.current() != states.next()) {
            switch (states.next()) {
                case FENCED:
                case SHUTDOWN_NOW:
                    handleBrokerFenced(brokerId, records);
                    break;
                case UNFENCED:
                    handleBrokerUnfenced(brokerId, brokerEpoch, records);
                    break;
                case CONTROLLED_SHUTDOWN:
                    handleBrokerInControlledShutdown(brokerId, brokerEpoch, records);
                    break;
            }
        }
        heartbeatManager.touch(brokerId,
            states.next().fenced(),
            request.currentMetadataOffset());
        if (featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
            handleDirectoriesOffline(brokerId, brokerEpoch, request.offlineLogDirs(), records);
        }
        boolean isCaughtUp = request.currentMetadataOffset() >= registerBrokerRecordOffset;
        BrokerHeartbeatReply reply = new BrokerHeartbeatReply(isCaughtUp,
                states.next().fenced(),
                states.next().inControlledShutdown(),
                states.next().shouldShutDown());
        return ControllerResult.of(records, reply);
    }
Generates the appropriate records to handle a list of directories being reported offline.

If the reported directories include directories that were previously online, this includes
a BrokerRegistrationChangeRecord and any number of PartitionChangeRecord to update
leadership and ISR for partitions in those directories that were previously online.

@param brokerId    The broker id.
@param brokerEpoch The broker epoch.
@param offlineDirs The list of directories that are offline.
@param records     The record list to append to.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java