@Test public void testUpdateAndCreateTopic() throws Exception { 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."); } 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); }
private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); final CreateTopicRequestHeader requestHeader = (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class); log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); String topic = requestHeader.getTopic(); if (!TopicValidator.validateTopic(topic, response)) { return response; } if (TopicValidator.isSystemTopic(topic, response)) { 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()); this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig); this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion()); response.setCode(ResponseCode.SUCCESS); return response; }
AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor { private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); final CreateTopicRequestHeader requestHeader = (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class); log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); String topic = requestHeader.getTopic(); if (!TopicValidator.validateTopic(topic, response)) { return response; } if (TopicValidator.isSystemTopic(topic, response)) { 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()); this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig); this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion()); response.setCode(ResponseCode.SUCCESS); return response; } }
AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor { private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); final CreateTopicRequestHeader requestHeader = (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class); log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); String topic = requestHeader.getTopic(); if (!TopicValidator.validateTopic(topic, response)) { return response; } if (TopicValidator.isSystemTopic(topic, response)) { 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()); this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig); this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion()); response.setCode(ResponseCode.SUCCESS); return response; } AdminBrokerProcessor(final BrokerController brokerController); }
AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor { private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); final CreateTopicRequestHeader requestHeader = (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class); log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); String topic = requestHeader.getTopic(); if (!TopicValidator.validateTopic(topic, response)) { return response; } if (TopicValidator.isSystemTopic(topic, response)) { 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()); this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig); this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion()); response.setCode(ResponseCode.SUCCESS); return response; } AdminBrokerProcessor(final BrokerController brokerController); @Override RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request); @Override boolean rejectRequest(); RemotingCommand resetOffset(ChannelHandlerContext ctx,
        RemotingCommand request); RemotingCommand getConsumerStatus(ChannelHandlerContext ctx,
        RemotingCommand request); RemotingCommand cleanExpiredConsumeQueue(); RemotingCommand cleanUnusedTopic(); }
AdminBrokerProcessor extends AsyncNettyRequestProcessor implements NettyRequestProcessor { private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException { final RemotingCommand response = RemotingCommand.createResponseCommand(null); final CreateTopicRequestHeader requestHeader = (CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class); log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel())); String topic = requestHeader.getTopic(); if (!TopicValidator.validateTopic(topic, response)) { return response; } if (TopicValidator.isSystemTopic(topic, response)) { 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()); this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig); this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion()); response.setCode(ResponseCode.SUCCESS); return response; } AdminBrokerProcessor(final BrokerController brokerController); @Override RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request); @Override boolean rejectRequest(); RemotingCommand resetOffset(ChannelHandlerContext ctx,
        RemotingCommand request); RemotingCommand getConsumerStatus(ChannelHandlerContext ctx,
        RemotingCommand request); RemotingCommand cleanExpiredConsumeQueue(); RemotingCommand cleanUnusedTopic();  }