ControllerResult<Void> maybeFenceOneStaleBroker() {
        List<ApiMessageAndVersion> records = new ArrayList<>();
        BrokerHeartbeatManager heartbeatManager = clusterControl.heartbeatManager();
        heartbeatManager.findOneStaleBroker().ifPresent(brokerId -> {
            // Even though multiple brokers can go stale at a time, we will process
            // fencing one at a time so that the effect of fencing each broker is visible
            // to the system prior to processing the next one
            log.info("Fencing broker {} because its session has timed out.", brokerId);
            handleBrokerFenced(brokerId, records);
            heartbeatManager.fence(brokerId);
        });
        return ControllerResult.of(records, null);
    }
Process a broker heartbeat which has been sitting on the queue for too long, and has
expired. With default settings, this would happen after 1 second. We process expired
heartbeats by updating the lastSeenNs of the broker, so that the broker won't get fenced
incorrectly. However, we don't perform any state changes that we normally would, such as
unfencing a fenced broker, etc.
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java