public Controller waitForActiveController() throws InterruptedException {
        AtomicReference<Controller> active = new AtomicReference<>(null);
        TestUtils.retryOnExceptionWithTimeout(60_000, () -> {
            for (ControllerServer controllerServer : controllers.values()) {
                if (controllerServer.controller().isActive()) {
                    active.set(controllerServer.controller());
                }
            }
            assertNotNull(active.get(), "No active controller found");
        });
        return active.get();
    }
Wait for a controller to mark all the brokers as ready (registered and unfenced).
And also wait for the metadata cache up-to-date in each broker server.
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java