private synchronized RemotingCommand updateAndCreateTopicList(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        long startTime = System.currentTimeMillis();

        final CreateTopicListRequestBody requestBody = CreateTopicListRequestBody.decode(request.getBody(), CreateTopicListRequestBody.class);
        List<TopicConfig> topicConfigList = requestBody.getTopicConfigList();

        StringBuilder builder = new StringBuilder();
        for (TopicConfig topicConfig : topicConfigList) {
            builder.append(topicConfig.getTopicName()).append(";");
        }
        String topicNames = builder.toString();
        LOGGER.info("AdminBrokerProcessor#updateAndCreateTopicList: topicNames: {}, called by {}", topicNames, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        final RemotingCommand response = RemotingCommand.createResponseCommand(null);

        long executionTime;

        try {
            // Valid topics
            for (TopicConfig topicConfig : topicConfigList) {
                String topic = topicConfig.getTopicName();
                TopicValidator.ValidateTopicResult result = TopicValidator.validateTopic(topic);
                if (!result.isValid()) {
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark(result.getRemark());
                    return response;
                }
                if (brokerController.getBrokerConfig().isValidateSystemTopicWhenUpdateTopic()) {
                    if (TopicValidator.isSystemTopic(topic)) {
                        response.setCode(ResponseCode.SYSTEM_ERROR);
                        response.setRemark("The topic[" + topic + "] is conflict with system topic.");
                        return response;
                    }
                }
                if (topicConfig.getTopicMessageType() == TopicMessageType.MIXED
                    && !brokerController.getBrokerConfig().isEnableMixedMessageType()) {
                    response.setCode(ResponseCode.SYSTEM_ERROR);
                    response.setRemark("MIXED message type is not supported.");
                    return response;
                }
                if (topicConfig.equals(this.brokerController.getTopicConfigManager().getTopicConfigTable().get(topic))) {
                    LOGGER.info("Broker receive request to update or create topic={}, but topicConfig has  no changes , so idempotent, caller address={}",
                        topic, RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
                    response.setCode(ResponseCode.SUCCESS);
                    return response;
                }
            }

            this.brokerController.getTopicConfigManager().updateTopicConfigList(topicConfigList);
            if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
                for (TopicConfig topicConfig : topicConfigList) {
                    this.brokerController.registerSingleTopicAll(topicConfig);
                }
            } else {
                this.brokerController.registerIncrementBrokerData(topicConfigList, this.brokerController.getTopicConfigManager().getDataVersion());
            }
            response.setCode(ResponseCode.SUCCESS);
        } catch (Exception e) {
            LOGGER.error("Update / create topic failed for [{}]", request, e);
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(e.getMessage());
            return response;
        }
        finally {
            executionTime = System.currentTimeMillis() - startTime;
            InvocationStatus status = response.getCode() == ResponseCode.SUCCESS ?
                InvocationStatus.SUCCESS : InvocationStatus.FAILURE;
            Attributes attributes = BrokerMetricsManager.newAttributesBuilder()
                .put(LABEL_INVOCATION_STATUS, status.getName())
                .put(LABEL_IS_SYSTEM, TopicValidator.isSystemTopic(topicNames))
                .build();
            BrokerMetricsManager.topicCreateExecuteTime.record(executionTime, attributes);
        }
        LOGGER.info("executionTime of all topics:{} is {} ms" , topicNames, executionTime);
        return response;
    }
@Test
    public void testUpdateAndCreateTopicList() throws RemotingCommandException {
        List<String> systemTopicList = new ArrayList<>(systemTopicSet);
        RemotingCommand request = buildCreateTopicListRequest(systemTopicList);
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
        assertThat(response.getRemark()).isEqualTo("The topic[" + systemTopicList.get(0) + "] is conflict with system topic.");

        List<String> inValidTopicList = new ArrayList<>();
        inValidTopicList.add("");
        request = buildCreateTopicListRequest(inValidTopicList);
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);

        List<String> topicList = new ArrayList<>();
        topicList.add("TEST_CREATE_TOPIC");
        topicList.add("TEST_CREATE_TOPIC1");
        request = buildCreateTopicListRequest(topicList);
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
        //test no changes
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }