class ConsumerGroupTopicServiceImpl extends AbstractBaseService<ConsumerGroupTopicEntity> implements CacheUpdateService, ConsumerGroupTopicService, TimerService {

	protected ConsumerGroupTopicCreateResponse createConsumerGroupTopicAndFailTopic(
            ConsumerGroupTopicCreateRequest consumerGroupTopicCreateRequest, Map<String, ConsumerGroupEntity> consumerGroupMap) {
		CacheUpdateHelper.updateCache();
		if (roleService.getRole(userInfoHolder.getUserId(), consumerGroupMap
				.get(consumerGroupTopicCreateRequest.getConsumerGroupName()).getOwnerIds()) >= UserRoleEnum.USER
				.getRoleCode()) {
			throw new RuntimeException(userInfoHolder.getUserId() + "没有操作权限，请进行权限检查。");
		}
		// 创建正常topic对应的consumerGroupTopic
		if (StringUtils.isEmpty(consumerGroupTopicCreateRequest.getTopicName())) {
			return new ConsumerGroupTopicCreateResponse("1", "主题不能为空");
		}
		ConsumerGroupTopicEntity consumerGroupTopicEntity = createConsumerGroupTopic(consumerGroupTopicCreateRequest);
		ConsumerGroupEntity consumerGroupEntity = consumerGroupService
				.get(consumerGroupTopicCreateRequest.getConsumerGroupId());
		TopicEntity topicEntity = topicService.get(consumerGroupTopicCreateRequest.getTopicId());
		TopicEntity failTopicEntity = topicService.createFailTopic(topicEntity, consumerGroupEntity);
		// 创建失败topic的consumerGroupTopic
		consumerGroupTopicCreateRequest.setTopicId(failTopicEntity.getId());
		consumerGroupTopicCreateRequest.setTopicName(failTopicEntity.getName());
		consumerGroupTopicCreateRequest.setTopicType(failTopicEntity.getTopicType());
		consumerGroupTopicCreateRequest.setDelayProcessTime(soaConfig.getFailTopicDelayProcessTime());
		consumerGroupTopicCreateRequest.setThreadSize(soaConfig.getFailTopicThreadSize());
		ConsumerGroupTopicEntity failConsumerGroupTopicEntity = createConsumerGroupTopic(
                consumerGroupTopicCreateRequest);

		ConsumerGroupTopicCreateResponse consumerGroupTopicCreateResponse=new ConsumerGroupTopicCreateResponse();
		try{
			// 创建正常topic对应的queueOffset
			queueOffsetService.createQueueOffset(consumerGroupTopicEntity);
			// 创建失败topic的queueOffset
			queueOffsetService.createQueueOffset(failConsumerGroupTopicEntity);
			consumerGroupService.addTopicNameToConsumerGroup(consumerGroupTopicEntity);
			consumerGroupService.notifyMeta(consumerGroupTopicCreateRequest.getConsumerGroupId());
			consumerGroupService.notifyRb(consumerGroupTopicCreateRequest.getConsumerGroupId());
		}catch (Exception e){
			consumerGroupTopicCreateResponse.setMsg(e.getMessage());
			consumerGroupTopicCreateResponse.setCode("1");
			throw new RuntimeException(e);
		}

		return consumerGroupTopicCreateResponse;
	}

}