public void rollingBrokerRestart(Optional<ClusterConfig> clusterConfig) {
            requireNonNull(clusterConfig);
            if (!started.get()) {
                throw new IllegalStateException("Tried to restart brokers but the cluster has not been started!");
            }
            for (int i = 0; i < clusterReference.get().brokerCount(); i++) {
                clusterReference.get().killBroker(i);
            }
            clusterConfig.ifPresent(config -> clusterReference.get().setClusterConfig(config));
            clusterReference.get().restartDeadBrokers(true);
            clusterReference.get().adminClientConfig().put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
        }
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