static void sendBrokerHeartbeatToUnfenceBrokers(
        QuorumController controller,
        List<Integer> brokers,
        Map<Integer, Long> brokerEpochs
    ) throws Exception {
        if (brokers.isEmpty()) {
            return;
        }
        for (Integer brokerId : brokers) {
            BrokerHeartbeatReply reply = controller.processBrokerHeartbeat(ANONYMOUS_CONTEXT,
                new BrokerHeartbeatRequestData()
                    .setWantFence(false)
                    .setBrokerEpoch(brokerEpochs.get(brokerId))
                    .setBrokerId(brokerId)
                    .setCurrentMetadataOffset(100000)
            ).get();
            assertEquals(new BrokerHeartbeatReply(true, false, false, false), reply);
        }
    }
Send broker heartbeats for the provided brokers.

@param controller    The active controller.
@param brokers       The broker IDs to send heartbeats for.
@param brokerEpochs  A map from broker ID to broker epoch.
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerIntegrationTestUtils.java