ControllerResult<CreateTopicsResponseData> createTopics(
        ControllerRequestContext context,
        CreateTopicsRequestData request,
        Set<String> describable
    ) {
        Map<String, ApiError> topicErrors = new HashMap<>();
        List<ApiMessageAndVersion> records = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);

        // Check the topic names.
        validateNewTopicNames(topicErrors, request.topics(), topicsWithCollisionChars);

        // Identify topics that already exist and mark them with the appropriate error
        request.topics().stream().filter(creatableTopic -> topicsByName.containsKey(creatableTopic.name()))
                .forEach(t -> topicErrors.put(t.name(), new ApiError(Errors.TOPIC_ALREADY_EXISTS,
                    "Topic '" + t.name() + "' already exists.")));

        // Verify that the configurations for the new topics are OK, and figure out what
        // configurations should be created.
        Map<ConfigResource, Map<String, Entry<OpType, String>>> configChanges =
            computeConfigChanges(topicErrors, request.topics());

        // Try to create whatever topics are needed.
        Map<String, CreatableTopicResult> successes = new HashMap<>();
        for (CreatableTopic topic : request.topics()) {
            if (topicErrors.containsKey(topic.name())) continue;
            // Figure out what ConfigRecords should be created, if any.
            ConfigResource configResource = new ConfigResource(TOPIC, topic.name());
            Map<String, Entry<OpType, String>> keyToOps = configChanges.get(configResource);
            List<ApiMessageAndVersion> configRecords;
            if (keyToOps != null) {
                ControllerResult<ApiError> configResult =
                    configurationControl.incrementalAlterConfig(configResource, keyToOps, true);
                if (configResult.response().isFailure()) {
                    topicErrors.put(topic.name(), configResult.response());
                    continue;
                } else {
                    configRecords = configResult.records();
                }
            } else {
                configRecords = Collections.emptyList();
            }
            ApiError error;
            try {
                error = createTopic(context, topic, records, successes, configRecords, describable.contains(topic.name()));
            } catch (ApiException e) {
                error = ApiError.fromThrowable(e);
            }
            if (error.isFailure()) {
                topicErrors.put(topic.name(), error);
            }
        }

        // Create responses for all topics.
        CreateTopicsResponseData data = new CreateTopicsResponseData();
        StringBuilder resultsBuilder = new StringBuilder();
        String resultsPrefix = "";
        for (CreatableTopic topic : request.topics()) {
            ApiError error = topicErrors.get(topic.name());
            if (error != null) {
                data.topics().add(new CreatableTopicResult().
                    setName(topic.name()).
                    setErrorCode(error.error().code()).
                    setErrorMessage(error.message()));
                resultsBuilder.append(resultsPrefix).append(topic).append(": ").
                    append(error.error()).append(" (").append(error.message()).append(")");
                resultsPrefix = ", ";
                continue;
            }
            CreatableTopicResult result = successes.get(topic.name());
            data.topics().add(result);
            resultsBuilder.append(resultsPrefix).append(topic).append(": ").
                append("SUCCESS");
            resultsPrefix = ", ";
        }
        if (request.validateOnly()) {
            log.info("Validate-only CreateTopics result(s): {}", resultsBuilder);
            return ControllerResult.atomicOf(Collections.emptyList(), data);
        } else {
            log.info("CreateTopics result(s): {}", resultsBuilder);
            return ControllerResult.atomicOf(records, data);
        }
    }
@Test
    public void testCreateTopics() {
        ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
        ReplicationControlManager replicationControl = ctx.replicationControl;
        CreateTopicsRequestData request = new CreateTopicsRequestData();
        request.topics().add(new CreatableTopic().setName("foo").
            setNumPartitions(-1).setReplicationFactor((short) -1));

        ControllerRequestContext requestContext = anonymousContextFor(ApiKeys.CREATE_TOPICS);
        ControllerResult<CreateTopicsResponseData> result =
            replicationControl.createTopics(requestContext, request, Collections.singleton("foo"));
        CreateTopicsResponseData expectedResponse = new CreateTopicsResponseData();
        expectedResponse.topics().add(new CreatableTopicResult().setName("foo").
            setErrorCode(INVALID_REPLICATION_FACTOR.code()).
                setErrorMessage("Unable to replicate the partition 3 time(s): All " +
                    "brokers are currently fenced."));
        assertEquals(expectedResponse, result.response());

        ctx.registerBrokers(0, 1, 2);
        ctx.unfenceBrokers(0);
        ctx.inControlledShutdownBrokers(0);

        ControllerResult<CreateTopicsResponseData> result2 =
            replicationControl.createTopics(requestContext, request, Collections.singleton("foo"));
        CreateTopicsResponseData expectedResponse2 = new CreateTopicsResponseData();
        expectedResponse2.topics().add(new CreatableTopicResult().setName("foo").
            setErrorCode(INVALID_REPLICATION_FACTOR.code()).
            setErrorMessage("Unable to replicate the partition 3 time(s): All " +
                "brokers are currently fenced or in controlled shutdown."));
        assertEquals(expectedResponse2, result2.response());

        ctx.registerBrokers(0, 1, 2);
        ctx.unfenceBrokers(0, 1, 2);

        ControllerResult<CreateTopicsResponseData> result3 =
            replicationControl.createTopics(requestContext, request, Collections.singleton("foo"));
        CreateTopicsResponseData expectedResponse3 = new CreateTopicsResponseData();
        expectedResponse3.topics().add(new CreatableTopicResult().setName("foo").
            setNumPartitions(1).setReplicationFactor((short) 3).
            setErrorMessage(null).setErrorCode((short) 0).
            setTopicId(result3.response().topics().find("foo").topicId()));
        assertEquals(expectedResponse3, result3.response());
        ctx.replay(result3.records());
        assertEquals(new PartitionRegistration.Builder().setReplicas(new int[] {1, 2, 0}).
            setDirectories(new Uuid[] {
                    Uuid.fromString("TESTBROKER00001DIRAAAA"),
                    Uuid.fromString("TESTBROKER00002DIRAAAA"),
                    Uuid.fromString("TESTBROKER00000DIRAAAA")
            }).
            setIsr(new int[] {1, 2, 0}).setLeader(1).setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).setLeaderEpoch(0).setPartitionEpoch(0).build(),
            replicationControl.getPartition(
                ((TopicRecord) result3.records().get(0).message()).topicId(), 0));
        ControllerResult<CreateTopicsResponseData> result4 =
                replicationControl.createTopics(requestContext, request, Collections.singleton("foo"));
        CreateTopicsResponseData expectedResponse4 = new CreateTopicsResponseData();
        expectedResponse4.topics().add(new CreatableTopicResult().setName("foo").
                setErrorCode(Errors.TOPIC_ALREADY_EXISTS.code()).
                setErrorMessage("Topic 'foo' already exists."));
        assertEquals(expectedResponse4, result4.response());
    }