BrokerControlStates calculateNextBrokerState(int brokerId,
                                                 BrokerHeartbeatRequestData request,
                                                 long registerBrokerRecordOffset,
                                                 Supplier<Boolean> hasLeaderships) {
        BrokerHeartbeatState broker = heartbeatStateOrThrow(brokerId);
        BrokerControlState currentState = currentBrokerState(broker);
        switch (currentState) {
            case FENCED:
                if (request.wantShutDown()) {
                    log.info("Fenced broker {} has requested and been granted an immediate " +
                        "shutdown.", brokerId);
                    return new BrokerControlStates(currentState, SHUTDOWN_NOW);
                } else if (!request.wantFence()) {
                    if (request.currentMetadataOffset() >= registerBrokerRecordOffset) {
                        log.info("The request from broker {} to unfence has been granted " +
                                "because it has caught up with the offset of its register " +
                                "broker record {}.", brokerId, registerBrokerRecordOffset);
                        return new BrokerControlStates(currentState, UNFENCED);
                    } else {
                        if (log.isDebugEnabled()) {
                            log.debug("The request from broker {} to unfence cannot yet " +
                                "be granted because it has not caught up with the offset of " +
                                "its register broker record {}. It is still at offset {}.",
                                brokerId, registerBrokerRecordOffset, request.currentMetadataOffset());
                        }
                        return new BrokerControlStates(currentState, FENCED);
                    }
                }
                return new BrokerControlStates(currentState, FENCED);

            case UNFENCED:
                if (request.wantFence()) {
                    if (request.wantShutDown()) {
                        log.info("Unfenced broker {} has requested and been granted an " +
                            "immediate shutdown.", brokerId);
                        return new BrokerControlStates(currentState, SHUTDOWN_NOW);
                    } else {
                        log.info("Unfenced broker {} has requested and been granted " +
                            "fencing", brokerId);
                        return new BrokerControlStates(currentState, FENCED);
                    }
                } else if (request.wantShutDown()) {
                    if (hasLeaderships.get()) {
                        log.info("Unfenced broker {} has requested and been granted a " +
                            "controlled shutdown.", brokerId);
                        return new BrokerControlStates(currentState, CONTROLLED_SHUTDOWN);
                    } else {
                        log.info("Unfenced broker {} has requested and been granted an " +
                            "immediate shutdown.", brokerId);
                        return new BrokerControlStates(currentState, SHUTDOWN_NOW);
                    }
                }
                return new BrokerControlStates(currentState, UNFENCED);

            case CONTROLLED_SHUTDOWN:
                if (hasLeaderships.get()) {
                    log.debug("Broker {} is in controlled shutdown state, but can not " +
                        "shut down because more leaders still need to be moved.", brokerId);
                    return new BrokerControlStates(currentState, CONTROLLED_SHUTDOWN);
                }
                long lowestActiveOffset = lowestActiveOffset();
                if (broker.controlledShutdownOffset <= lowestActiveOffset) {
                    log.info("The request from broker {} to shut down has been granted " +
                        "since the lowest active offset {} is now greater than the " +
                        "broker's controlled shutdown offset {}.", brokerId,
                        lowestActiveOffset, broker.controlledShutdownOffset);
                    return new BrokerControlStates(currentState, SHUTDOWN_NOW);
                }
                log.debug("The request from broker {} to shut down can not yet be granted " +
                    "because the lowest active offset {} is not greater than the broker's " +
                    "shutdown offset {}.", brokerId, lowestActiveOffset,
                    broker.controlledShutdownOffset);
                return new BrokerControlStates(currentState, CONTROLLED_SHUTDOWN);

            default:
                return new BrokerControlStates(currentState, SHUTDOWN_NOW);
        }
    }
@Test
    public void testCalculateNextBrokerState() {
        BrokerHeartbeatManager manager = newBrokerHeartbeatManager();
        for (int brokerId = 0; brokerId < 6; brokerId++) {
            manager.register(brokerId, true);
        }
        manager.touch(0, true, 100);
        manager.touch(1, false, 98);
        manager.touch(2, false, 100);
        manager.touch(3, false, 100);
        manager.touch(4, true, 100);
        manager.touch(5, false, 99);
        manager.maybeUpdateControlledShutdownOffset(5, 99);

        assertEquals(98L, manager.lowestActiveOffset());

        assertEquals(new BrokerControlStates(FENCED, SHUTDOWN_NOW),
            manager.calculateNextBrokerState(0,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> false));
        assertEquals(new BrokerControlStates(FENCED, UNFENCED),
            manager.calculateNextBrokerState(0,
                new BrokerHeartbeatRequestData().setWantFence(false).
                    setCurrentMetadataOffset(100), 100, () -> false));
        assertEquals(new BrokerControlStates(FENCED, FENCED),
            manager.calculateNextBrokerState(0,
                new BrokerHeartbeatRequestData().setWantFence(false).
                    setCurrentMetadataOffset(50), 100, () -> false));
        assertEquals(new BrokerControlStates(FENCED, FENCED),
            manager.calculateNextBrokerState(0,
                new BrokerHeartbeatRequestData().setWantFence(true), 100, () -> false));

        assertEquals(new BrokerControlStates(UNFENCED, CONTROLLED_SHUTDOWN),
            manager.calculateNextBrokerState(1,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> true));
        assertEquals(new BrokerControlStates(UNFENCED, SHUTDOWN_NOW),
            manager.calculateNextBrokerState(1,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> false));
        assertEquals(new BrokerControlStates(UNFENCED, UNFENCED),
            manager.calculateNextBrokerState(1,
                new BrokerHeartbeatRequestData().setWantFence(false), 100, () -> false));

        assertEquals(new BrokerControlStates(CONTROLLED_SHUTDOWN, CONTROLLED_SHUTDOWN),
            manager.calculateNextBrokerState(5,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> true));
        assertEquals(new BrokerControlStates(CONTROLLED_SHUTDOWN, CONTROLLED_SHUTDOWN),
            manager.calculateNextBrokerState(5,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> false));
        manager.fence(1);
        assertEquals(new BrokerControlStates(CONTROLLED_SHUTDOWN, SHUTDOWN_NOW),
            manager.calculateNextBrokerState(5,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> false));
        assertEquals(new BrokerControlStates(CONTROLLED_SHUTDOWN, CONTROLLED_SHUTDOWN),
            manager.calculateNextBrokerState(5,
                new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> true));
        assertEquals("Broker 6 is not registered.",
                assertThrows(IllegalStateException.class,
                        () -> manager.calculateNextBrokerState(6, new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> true)).getMessage());
        assertEquals("Broker 7 is not registered.",
                assertThrows(IllegalStateException.class,
                        () -> manager.calculateNextBrokerState(7, new BrokerHeartbeatRequestData().setWantShutDown(true), 100, () -> true)).getMessage());
    }