private KafkaServer findBrokerOrThrow(int brokerId) {
            return brokers().values().stream()
                .filter(server -> server.config().brokerId() == brokerId)
                .map(s -> (KafkaServer) s)
                .findFirst()
                .orElseThrow(() -> new IllegalArgumentException("Unknown brokerId " + brokerId));
        }
Restart brokers with given cluster config.

@param clusterConfig clusterConfig is optional. If left Optional.empty(), brokers will restart without
                     reconfiguring configurations. Otherwise, the restart will reconfigure configurations
                     according to the provided cluster config.
core/src/test/java/kafka/test/junit/ZkClusterInvocationContext.java