private void formatNode(
        MetaPropertiesEnsemble ensemble,
        boolean writeMetadataDirectory
    ) {
        try {
            MetaPropertiesEnsemble.Copier copier =
                new MetaPropertiesEnsemble.Copier(MetaPropertiesEnsemble.EMPTY);
            for (Entry<String, MetaProperties> entry : ensemble.logDirProps().entrySet()) {
                String logDir = entry.getKey();
                if (writeMetadataDirectory || (!ensemble.metadataLogDir().equals(Optional.of(logDir)))) {
                    log.trace("Adding {} to the list of directories to format.", logDir);
                    copier.setLogDirProps(logDir, entry.getValue());
                }
            }
            copier.setPreWriteHandler((logDir, isNew, metaProperties) -> {
                log.info("Formatting {}.", logDir);
                Files.createDirectories(Paths.get(logDir));
                BootstrapDirectory bootstrapDirectory = new BootstrapDirectory(logDir, Optional.empty());
                bootstrapDirectory.writeBinaryFile(nodes.bootstrapMetadata());
            });
            copier.writeLogDirChanges();
        } catch (Exception e) {
            throw new RuntimeException("Failed to format node " + ensemble.nodeId(), e);
        }
    }
This class manages a future which is completed with the proper value for
controller.quorum.voters once the randomly assigned ports for all the controllers are
known.
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java