@Override
    @TpsControl(pointName = "ConfigPublish")
    @Secured(action = ActionTypes.WRITE, signType = SignType.CONFIG)
    @ExtractorManager.Extractor(rpcExtractor = ConfigRequestParamExtractor.class)
    public ConfigPublishResponse handle(ConfigPublishRequest request, RequestMeta meta) throws NacosException {
        
        try {
            String dataId = request.getDataId();
            String group = request.getGroup();
            String content = request.getContent();
            final String tenant = request.getTenant();
            
            final String srcIp = meta.getClientIp();
            final String requestIpApp = request.getAdditionParam("requestIpApp");
            final String tag = request.getAdditionParam("tag");
            final String appName = request.getAdditionParam("appName");
            final String type = request.getAdditionParam("type");
            final String srcUser = request.getAdditionParam("src_user");
            final String encryptedDataKey = request.getAdditionParam("encryptedDataKey");
            
            // check tenant
            ParamUtils.checkParam(dataId, group, "datumId", content);
            ParamUtils.checkParam(tag);
            Map<String, Object> configAdvanceInfo = new HashMap<>(10);
            MapUtil.putIfValNoNull(configAdvanceInfo, "config_tags", request.getAdditionParam("config_tags"));
            MapUtil.putIfValNoNull(configAdvanceInfo, "desc", request.getAdditionParam("desc"));
            MapUtil.putIfValNoNull(configAdvanceInfo, "use", request.getAdditionParam("use"));
            MapUtil.putIfValNoNull(configAdvanceInfo, "effect", request.getAdditionParam("effect"));
            MapUtil.putIfValNoNull(configAdvanceInfo, "type", type);
            MapUtil.putIfValNoNull(configAdvanceInfo, "schema", request.getAdditionParam("schema"));
            ParamUtils.checkParam(configAdvanceInfo);
            
            if (AggrWhitelist.isAggrDataId(dataId)) {
                Loggers.REMOTE_DIGEST.warn("[aggr-conflict] {} attempt to publish single data, {}, {}", srcIp, dataId,
                        group);
                throw new NacosException(NacosException.NO_RIGHT, "dataId:" + dataId + " is aggr");
            }
            
            ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);
            configInfo.setMd5(request.getCasMd5());
            configInfo.setType(type);
            configInfo.setEncryptedDataKey(encryptedDataKey);
            String betaIps = request.getAdditionParam("betaIps");
            ConfigOperateResult configOperateResult = null;
            String persistEvent = ConfigTraceService.PERSISTENCE_EVENT;
            if (StringUtils.isBlank(betaIps)) {
                if (StringUtils.isBlank(tag)) {
                    if (StringUtils.isNotBlank(request.getCasMd5())) {
                        configOperateResult = configInfoPersistService.insertOrUpdateCas(srcIp, srcUser, configInfo,
                                configAdvanceInfo);
                        if (!configOperateResult.isSuccess()) {
                            return ConfigPublishResponse.buildFailResponse(ResponseCode.FAIL.getCode(),
                                    "Cas publish fail,server md5 may have changed.");
                        }
                    } else {
                        configOperateResult = configInfoPersistService.insertOrUpdate(srcIp, srcUser, configInfo,
                                configAdvanceInfo);
                    }
                    ConfigChangePublisher.notifyConfigChange(new ConfigDataChangeEvent(false, dataId, group, tenant,
                            configOperateResult.getLastModified()));
                } else {
                    if (StringUtils.isNotBlank(request.getCasMd5())) {
                        configOperateResult = configInfoTagPersistService.insertOrUpdateTagCas(configInfo, tag, srcIp,
                                srcUser);
                        if (!configOperateResult.isSuccess()) {
                            return ConfigPublishResponse.buildFailResponse(ResponseCode.FAIL.getCode(),
                                    "Cas publish tag config fail,server md5 may have changed.");
                        }
                    } else {
                        configOperateResult = configInfoTagPersistService.insertOrUpdateTag(configInfo, tag, srcIp,
                                srcUser);
                    }
                    persistEvent = ConfigTraceService.PERSISTENCE_EVENT_TAG + "-" + tag;
                    ConfigChangePublisher.notifyConfigChange(
                            new ConfigDataChangeEvent(false, dataId, group, tenant, tag,
                                    configOperateResult.getLastModified()));
                }
            } else {
                // beta publish
                if (StringUtils.isNotBlank(request.getCasMd5())) {
                    configOperateResult = configInfoBetaPersistService.insertOrUpdateBetaCas(configInfo, betaIps, srcIp,
                            srcUser);
                    if (!configOperateResult.isSuccess()) {
                        return ConfigPublishResponse.buildFailResponse(ResponseCode.FAIL.getCode(),
                                "Cas publish beta config fail,server md5 may have changed.");
                    }
                } else {
                    configOperateResult = configInfoBetaPersistService.insertOrUpdateBeta(configInfo, betaIps, srcIp,
                            srcUser);
                }
                persistEvent = ConfigTraceService.PERSISTENCE_EVENT_BETA;
                
                ConfigChangePublisher.notifyConfigChange(
                        new ConfigDataChangeEvent(true, dataId, group, tenant, configOperateResult.getLastModified()));
            }
            ConfigTraceService.logPersistenceEvent(dataId, group, tenant, requestIpApp,
                    configOperateResult.getLastModified(), srcIp, persistEvent, ConfigTraceService.PERSISTENCE_TYPE_PUB,
                    content);
            return ConfigPublishResponse.buildSuccessResponse();
        } catch (Exception e) {
            Loggers.REMOTE_DIGEST.error("[ConfigPublishRequestHandler] publish config error ,request ={}", request, e);
            return ConfigPublishResponse.buildFailResponse(
                    (e instanceof NacosException) ? ((NacosException) e).getErrCode() : ResponseCode.FAIL.getCode(),
                    e.getMessage());
        }
    }
@Test
    void testBetaPublishCas() throws NacosException, InterruptedException {
        String dataId = "testBetaPublishCas";
        String group = "group";
        String tenant = "tenant";
        String content = "content";
        
        ConfigPublishRequest configPublishRequest = new ConfigPublishRequest();
        configPublishRequest.setDataId(dataId);
        configPublishRequest.setGroup(group);
        configPublishRequest.setTenant(tenant);
        configPublishRequest.setContent(content);
        configPublishRequest.setCasMd5("12314532");
        Map<String, String> keyMap = new HashMap<>();
        String srcUser = "src_user111";
        keyMap.put("src_user", srcUser);
        String betaIps = "127.0.0.1,127.0.0.2";
        keyMap.put("betaIps", betaIps);
        configPublishRequest.setAdditionMap(keyMap);
        
        RequestMeta requestMeta = new RequestMeta();
        String clientIp = "127.0.0.1";
        requestMeta.setClientIp(clientIp);
        
        AtomicReference<ConfigDataChangeEvent> reference = new AtomicReference<>();
        NotifyCenter.registerSubscriber(new Subscriber() {
            
            @Override
            public void onEvent(Event event) {
                ConfigDataChangeEvent event1 = (ConfigDataChangeEvent) event;
                if (event1.dataId.equals(dataId)) {
                    reference.set((ConfigDataChangeEvent) event);
                }
            }
            
            @Override
            public Class<? extends Event> subscribeType() {
                return ConfigDataChangeEvent.class;
            }
        });
        
        ConfigOperateResult configOperateResult = new ConfigOperateResult(true);
        long timestamp = System.currentTimeMillis();
        long id = timestamp / 1000;
        configOperateResult.setId(id);
        configOperateResult.setLastModified(timestamp);
        when(configInfoBetaPersistService.insertOrUpdateBetaCas(any(ConfigInfo.class), eq(betaIps), eq(requestMeta.getClientIp()),
                eq(srcUser))).thenReturn(configOperateResult);
        ConfigPublishResponse response = configPublishRequestHandler.handle(configPublishRequest, requestMeta);
        
        assertEquals(ResponseCode.SUCCESS.getCode(), response.getResultCode());
        Thread.sleep(500L);
        assertTrue(reference.get() != null);
        assertEquals(dataId, reference.get().dataId);
        assertEquals(group, reference.get().group);
        assertEquals(tenant, reference.get().tenant);
        assertEquals(timestamp, reference.get().lastModifiedTs);
        assertFalse(reference.get().isBatch);
        assertTrue(reference.get().isBeta);
        
    }