private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        long startTime = System.currentTimeMillis();
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final CreateTopicRequestHeader requestHeader =
            (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class);

        LOGGER.info("Broker receive request to update or create topic={}, caller address={}",
            requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        String topic = requestHeader.getTopic();

        long executionTime;
        try {
            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;
                }
            }

            TopicConfig topicConfig = new TopicConfig(topic);
            topicConfig.setReadQueueNums(requestHeader.getReadQueueNums());
            topicConfig.setWriteQueueNums(requestHeader.getWriteQueueNums());
            topicConfig.setTopicFilterType(requestHeader.getTopicFilterTypeEnum());
            topicConfig.setPerm(requestHeader.getPerm());
            topicConfig.setTopicSysFlag(requestHeader.getTopicSysFlag() == null ? 0 : requestHeader.getTopicSysFlag());
            topicConfig.setOrder(requestHeader.getOrder());
            String attributesModification = requestHeader.getAttributes();
            topicConfig.setAttributes(AttributeParser.parseToMap(attributesModification));

            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={}",
                    requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
                response.setCode(ResponseCode.SUCCESS);
                return response;
            }

            this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
            if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
                this.brokerController.registerSingleTopicAll(topicConfig);
            } else {
                this.brokerController.registerIncrementBrokerData(topicConfig, 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(topic))
                    .build();
            BrokerMetricsManager.topicCreateExecuteTime.record(executionTime, attributes);
        }
        LOGGER.info("executionTime of create topic:{} is {} ms" , topic, executionTime);
        return response;
    }
@Test
    public void testUpdateAndCreateTopic() throws Exception {
        //test system topic
        for (String topic : systemTopicSet) {
            RemotingCommand request = buildCreateTopicRequest(topic);
            RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
            assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);
            assertThat(response.getRemark()).isEqualTo("The topic[" + topic + "] is conflict with system topic.");
        }

        //test validate error topic
        String topic = "";
        RemotingCommand request = buildCreateTopicRequest(topic);
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);

        topic = "TEST_CREATE_TOPIC";
        request = buildCreateTopicRequest(topic);
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);
    }