Nacos学习手册

    <properties>
        <revision>3.2.4</revision>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <!-- Compiler settings properties -->
        <client.java.version>1.8</client.java.version>
        <java.version>17</java.version>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
        <!-- Maven properties -->
        <maven.test.skip>false</maven.test.skip>
        <maven.javadoc.skip>true</maven.javadoc.skip>
        <!-- Exclude all generated code -->
        <sonar.exclusions>file:**/generated-sources/**,**/test/**</sonar.exclusions>
        
        <!-- plugin version -->
        <versions-maven-plugin.version>2.2</versions-maven-plugin.version>
        <dependency-mediator-maven-plugin.version>1.0.2</dependency-mediator-maven-plugin.version>
        <clirr-maven-plugin.version>2.7</clirr-maven-plugin.version>
        <maven-enforcer-plugin.version>3.5.0</maven-enforcer-plugin.version>
        <maven-compiler-plugin.version>3.5.1</maven-compiler-plugin.version>
        <maven-javadoc-plugin.version>2.10.4</maven-javadoc-plugin.version>
        <maven-jar-plugin.version>3.2.2</maven-jar-plugin.version>
        <maven-source-plugin.version>3.0.1</maven-source-plugin.version>

        <apache-rat-plugin.version>0.12</apache-rat-plugin.version>
        <maven-resources-plugin.version>3.0.2</maven-resources-plugin.version>
        <jacoco-maven-plugin.version>0.8.13</jacoco-maven-plugin.version>
        <maven-surefire-plugin.version>3.2.5</maven-surefire-plugin.version>
        <spotbugs-maven-plugin.version>4.8.6.2</spotbugs-maven-plugin.version>
        <sonar-maven-plugin.version>3.0.2</sonar-maven-plugin.version>
        <maven-gpg-plugin.version>3.1.0</maven-gpg-plugin.version>
        <maven-failsafe-plugin.version>3.5.4</maven-failsafe-plugin.version>
        <maven-assembly-plugin.version>3.0.0</maven-assembly-plugin.version>
        <maven-checkstyle-plugin.version>3.6.0</maven-checkstyle-plugin.version>
        <maven-easyj-version>1.1.5</maven-easyj-version>
        <!-- dependency version related to plugin -->
        <extra-enforcer-rules.version>1.9.0</extra-enforcer-rules.version>
        
        <!-- dependency version -->
        <nacos.logback.adapter.version>1.1.5</nacos.logback.adapter.version>
        <spring-boot-dependencies.version>3.5.14</spring-boot-dependencies.version>
        <commons-io.version>2.14.0</commons-io.version>
        <commons-collections.version>3.2.2</commons-collections.version>
        <slf4j-api.version>2.0.13</slf4j-api.version>
        <logback.version>1.5.32</logback.version>
        <log4j.version>2.25.4</log4j.version>
        
        <mysql-connector-java.version>8.2.0</mysql-connector-java.version>
        <postgresql.version>42.7.11</postgresql.version>
        <derby.version>10.14.2.0</derby.version>
        <jjwt.version>0.11.2</jjwt.version>
        <grpc-java.version>1.78.0</grpc-java.version>
        <proto-google-common-protos.version>2.17.0</proto-google-common-protos.version>
        <protobuf-java.version>3.25.5</protobuf-java.version>
        <protoc-gen-grpc-java.version>${grpc-java.version}</protoc-gen-grpc-java.version>
        <hessian.version>4.0.63</hessian.version>

    /**
     * Default deployment type, means the nacos server and console are deployed in one process.
     */
    MERGED(Constants.NACOS_DEPLOYMENT_TYPE_MERGED),
    
    /**
     * Only server deployment type, means only the nacos server is deployed in the process.
     */
    SERVER(Constants.NACOS_DEPLOYMENT_TYPE_SERVER),
    
    /**
     * Only console deployment type, means only the nacos console is deployed in the process.
     */
    CONSOLE(Constants.NACOS_DEPLOYMENT_TYPE_CONSOLE),
    
    /**
     * Nacos Server and Mcp will be deployed in the process.
     */
    SERVER_WITH_MCP(Constants.NACOS_DEPLOYMENT_TYPE_SERVER_WITH_MCP),
    
    /**
     * Unknown deployment type.
     */
    ILLEGAL("unknown");
    
    private final String typeName;
    
    DeploymentType(String typeName) {
        this.typeName = typeName;
    }
    
    public String getTypeName() {
        return typeName;
    }
    
    public static DeploymentType getType(String type) {
        try {
            return DeploymentType.valueOf(type.toUpperCase());
        } catch (IllegalArgumentException e) {
            return ILLEGAL;
        }
    }
}

    public String getGroupedServiceName() {
        return NamingUtils.getGroupedName(name, group);
    }

    /**
     * register a new connect.
     *
     * @param connectionId connectionId
     * @param connection   connection
     */
    public synchronized boolean register(String connectionId, Connection connection) {
        
        if (connection.isConnected()) {
            String clientIp = connection.getMetaInfo().clientIp;
            if (connections.containsKey(connectionId)) {
                return true;
            }
            if (checkLimit(connection)) {
                return false;
            }
            if (traced(clientIp)) {
                connection.setTraced(true);
            }
            connections.put(connectionId, connection);
            connectionForClientIp.computeIfAbsent(clientIp, k -> new AtomicInteger(0))
                .getAndIncrement();
            
            clientConnectionEventListenerRegistry.notifyClientConnected(connection);
            
            LOGGER.info("new connection registered successfully, connectionId = {},connection={} ",
                connectionId,
                connection);
            return true;
            
        }
        return false;
        
    }

    /**
     * switch server .
     */
    protected void reconnect(final ServerInfo recommendServerInfo, boolean onRequestFail) {
        
        try {
            
            AtomicReference<ServerInfo> recommendServer =
                new AtomicReference<>(recommendServerInfo);
            if (onRequestFail && healthCheck()) {
                LoggerUtils.printIfInfoEnabled(LOGGER,
                    "[{}] Server check success, currentServer is {} ",
                    rpcClientConfig.name(), currentConnection.serverInfo.getAddress());
                rpcClientStatus.set(RpcClientStatus.RUNNING);
                return;
            }
            
            LoggerUtils.printIfInfoEnabled(LOGGER,
                "[{}] Try to reconnect to a new server, server is {}",
                rpcClientConfig.name(),
                recommendServerInfo == null ? " not appointed, will choose a random server."
                    : (recommendServerInfo.getAddress() + ", will try it once."));
            
            // loop until start client success.
            boolean switchSuccess = false;
            
            int reConnectTimes = 0;
            int retryTurns = 0;
            Exception lastException;
            while (!switchSuccess && !isShutdown()) {
                
                // 1.get a new server
                ServerInfo serverInfo = null;
                try {
                    serverInfo =
                        recommendServer.get() == null ? nextRpcServer() : recommendServer.get();
                    // 2.create a new channel to new server
                    Connection connectionNew = connectToServer(serverInfo);
                    if (connectionNew != null) {
                        LoggerUtils
                            .printIfInfoEnabled(LOGGER,
                                "[{}] Success to connect a server [{}], connectionId = {}",
                                rpcClientConfig.name(), serverInfo.getAddress(),
                                connectionNew.getConnectionId());
                        // successfully create a new connect.
                        if (currentConnection != null) {
                            LoggerUtils.printIfInfoEnabled(LOGGER,
                                "[{}] Abandon prev connection, server is {}, connectionId is {}",
                                rpcClientConfig.name(), currentConnection.serverInfo.getAddress(),
                                currentConnection.getConnectionId());
                            // set current connection to enable connection event.
                            currentConnection.setAbandon(true);
                            closeConnection(currentConnection);
                        }
                        currentConnection = connectionNew;
                        rpcClientStatus.set(RpcClientStatus.RUNNING);
                        switchSuccess = true;
                        eventLinkedBlockingQueue
                            .add(new ConnectionEvent(ConnectionEvent.CONNECTED, currentConnection));
                        return;
                    }
                    
                    // close connection if client is already shutdown.
                    if (isShutdown()) {
                        closeConnection(currentConnection);
                    }
                    
                    lastException = null;
                    
                } catch (Throwable throwable) {
                    LoggerUtils.printIfErrorEnabled(LOGGER, "Fail to connect server, error = {}",
                        throwable.getMessage());
                    lastException = new Exception(throwable);
                } finally {
                    recommendServer.set(null);
                }
                
                if (CollectionUtils.isEmpty(RpcClient.this.serverListFactory.getServerList())) {
                    throw new Exception("server list is empty");
                }
                
                if (reConnectTimes > 0
                    && reConnectTimes
                        % RpcClient.this.serverListFactory.getServerList().size() == 0) {
                    LoggerUtils.printIfInfoEnabled(LOGGER,
                        "[{}] Fail to connect server, after trying {} times, last try server is {}, error = {}",
                        rpcClientConfig.name(), reConnectTimes, serverInfo,
                        lastException == null ? "unknown" : lastException);
                    if (Integer.MAX_VALUE == retryTurns) {
                        retryTurns = 50;
                    } else {
                        retryTurns++;
                    }
                }
                
                reConnectTimes++;
                
                try {
                    // sleep x milliseconds to switch next server.
                    if (!isRunning()) {
                        // first round, try servers at a delay 100ms;second round, 200ms; max delays 5s. to be reconsidered.
                        Thread.sleep(Math.min(retryTurns + 1, 50) * 100L);
                    }
                } catch (InterruptedException e) {
                    // Do nothing.
                    // set the interrupted flag
                    Thread.currentThread().interrupt();
                }
            }
            
            if (isShutdown()) {
                LoggerUtils.printIfInfoEnabled(LOGGER,
                    "[{}] Client is shutdown, stop reconnect to server",
                    rpcClientConfig.name());
            }
            
        } catch (Exception e) {
            LoggerUtils
                .printIfWarnEnabled(LOGGER, "[{}] Fail to reconnect to server, error is {}",
                    rpcClientConfig.name(),
                    e);
        }
    }

    private InstanceResponse registerInstance(Service service, InstanceRequest request,
        RequestMeta meta)
        throws NacosException {
        Instance instance = request.getInstance();
        if (null == instance) {
            throw new NacosException(NacosException.INVALID_PARAM,
                "Required parameter 'instance' is missing.");
        }
        instance.validate();
        clientOperationService.registerInstance(service, instance, meta.getConnectionId());
        NotifyCenter.publishEvent(new RegisterInstanceTraceEvent(System.currentTimeMillis(),
            NamingRequestUtil.getSourceIpForGrpcRequest(meta), true, service.getNamespace(),
            service.getGroup(),
            service.getName(), instance.getIp(), instance.getPort()));
        return new InstanceResponse(NamingRemoteConstants.REGISTER_INSTANCE);
    }

    @Override
    public void registerInstance(Service service, Instance instance, String clientId)
        throws NacosException {
        NamingUtils.checkInstanceIsLegal(instance);
        
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        if (!singleton.isEphemeral()) {
            throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                String.format(
                    "Current service %s is persistent service, can't register ephemeral instance.",
                    singleton.getGroupedServiceName()));
        }
        Client client = clientManager.getClient(clientId);
        checkClientIsLegal(client, clientId);
        InstancePublishInfo instanceInfo = getPublishInfo(instance);
        client.addServiceInstance(singleton, instanceInfo);
        client.setLastUpdatedTime();
        client.recalculateRevision();
        NotifyCenter
            .publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
        NotifyCenter
            .publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton,
                instanceInfo.getMetadataId(), false));
    }

    private void handleClientOperation(ClientOperationEvent event) {
        Service service = event.getService();
        String clientId = event.getClientId();
        if (event instanceof ClientOperationEvent.ClientRegisterServiceEvent) {
            addPublisherIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) {
            removePublisherIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) {
            addSubscriberIndexes(service, clientId);
        } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) {
            removeSubscriberIndexes(service, clientId);
        }
    }

    @Override
    public Response requestReply(Request request, Connection connection) {
        if (request instanceof NotifySubscriberRequest) {
            NotifySubscriberRequest notifyRequest = (NotifySubscriberRequest) request;
            serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo());
            return new NotifySubscriberResponse();
        }
        return null;
    }

    /**
     * Process service info.
     *
     * @param serviceInfo new service info
     * @return service info
     */
    public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
        String serviceKey = serviceInfo.getKeyWithoutClusters();
        if (serviceKey == null) {
            NAMING_LOGGER.warn("process service info but serviceKey is null, service host: {}",
                JacksonUtils.toJson(serviceInfo.getHosts()));
            return null;
        }
        ServiceInfo oldService = serviceInfoMap.get(serviceKey);
        if (isEmptyOrErrorPush(serviceInfo)) {
            //empty or error push, just ignore
            NAMING_LOGGER.warn(
                "process service info but found empty or error push, serviceKey: {}, "
                    + "pushEmptyProtection: {}, hosts: {}",
                serviceKey, pushEmptyProtection, serviceInfo.getHosts());
            return oldService;
        }
        serviceInfoMap.put(serviceKey, serviceInfo);
        InstancesDiff diff = getServiceInfoDiff(oldService, serviceInfo);
        if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {
            serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));
        }
        
        if (enableClientMetrics) {
            try {
                MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
            } catch (Throwable t) {
                NAMING_LOGGER.error("Failed to update metrics for service info map size", t);
            }
        }
        
        if (diff.hasDifferent()) {
            NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(),
                serviceKey,
                JacksonUtils.toJson(serviceInfo.getHosts()));
            
            if (!failoverReactor.isFailoverSwitch(serviceKey)) {
                NotifyCenter.publishEvent(
                    new InstancesChangeEvent(notifierEventScope, serviceInfo.getName(),
                        serviceInfo.getGroupName(),
                        serviceInfo.getClusters(), serviceInfo.getHosts(), diff));
            }
            publishDiskCacheRefreshEvent(serviceKey, serviceInfo);
        }
        return serviceInfo;
    }

    @Override
    public boolean clientDisconnected(String clientId) {
        Loggers.SRV_LOG.info("Client connection {} disconnect, remove instances and subscribers",
            clientId);
        ConnectionBasedClient client = clients.remove(clientId);
        if (null == client) {
            return true;
        }
        client.release();
        boolean isResponsible = isResponsibleClient(client);
        NotifyCenter
            .publishEvent(new ClientOperationEvent.ClientReleaseEvent(client, isResponsible));
        NotifyCenter.publishEvent(new ClientEvent.ClientDisconnectEvent(client, isResponsible));
        return true;
    }

    @Override
    public void onDisConnect(Connection connection) {
        connected = false;
        LogUtils.NAMING_LOGGER.warn("Grpc connection disconnect, mark to redo");
        synchronized (registeredInstances) {
            registeredInstances.values()
                .forEach(instanceRedoData -> instanceRedoData.setRegistered(false));
        }
        synchronized (subscribes) {
            subscribes.values()
                .forEach(subscriberRedoData -> subscriberRedoData.setRegistered(false));
        }
        synchronized (namingFuzzyWatchServiceListHolder) {
            namingFuzzyWatchServiceListHolder.resetConsistenceStatus();
        }
        LogUtils.NAMING_LOGGER.warn("mark to redo completed");
    }

    private void syncToAllServer(ClientEvent event) {
        Client client = event.getClient();
        if (isInvalidClient(client)) {
            return;
        }
        if (event instanceof ClientEvent.ClientDisconnectEvent) {
            DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
            distroProtocol.sync(distroKey, DataOperation.DELETE);
        } else if (event instanceof ClientEvent.ClientChangedEvent) {
            DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
            distroProtocol.sync(distroKey, DataOperation.CHANGE);
        }
    }

    /**
     * Start to sync to target server.
     *
     * @param distroKey    distro key of sync data
     * @param action       the action of data operation
     * @param targetServer target server
     * @param delay        delay time for sync
     */
    public void syncToTarget(DistroKey distroKey, DataOperation action, String targetServer,
        long delay) {
        DistroKey distroKeyWithTarget =
            new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
                targetServer);
        DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
        distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget,
            distroDelayTask);
        if (Loggers.DISTRO.isDebugEnabled()) {
            Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, targetServer);
        }
    }

    private void upgradeClient(Client client, ClientSyncData clientSyncData) {
        Set<Service> syncedService = new HashSet<>();
        // process batch instance sync logic
        processBatchInstanceDistroData(syncedService, client, clientSyncData);
        List<String> namespaces = clientSyncData.getNamespaces();
        List<String> groupNames = clientSyncData.getGroupNames();
        List<String> serviceNames = clientSyncData.getServiceNames();
        List<InstancePublishInfo> instances = clientSyncData.getInstancePublishInfos();
        
        for (int i = 0; i < namespaces.size(); i++) {
            Service service =
                Service.newService(namespaces.get(i), groupNames.get(i), serviceNames.get(i));
            Service singleton = ServiceManager.getInstance().getSingleton(service);
            syncedService.add(singleton);
            InstancePublishInfo instancePublishInfo = instances.get(i);
            if (!instancePublishInfo.equals(client.getInstancePublishInfo(singleton))) {
                client.addServiceInstance(singleton, instancePublishInfo);
                NotifyCenter.publishEvent(
                    new ClientOperationEvent.ClientRegisterServiceEvent(singleton,
                        client.getClientId()));
                NotifyCenter.publishEvent(
                    new MetadataEvent.InstanceMetadataEvent(singleton,
                        instancePublishInfo.getMetadataId(), false));
            }
        }
        for (Service each : client.getAllPublishedService()) {
            if (!syncedService.contains(each)) {
                client.removeServiceInstance(each);
                NotifyCenter.publishEvent(
                    new ClientOperationEvent.ClientDeregisterServiceEvent(each,
                        client.getClientId()));
            }
        }
        client.setRevision(clientSyncData.getAttributes()
            .<Integer>getClientAttribute(ClientConstants.REVISION, 0));
    }

    private boolean loadAllDataSnapshotFromRemote(String resourceType) {
        DistroTransportAgent transportAgent =
            distroComponentHolder.findTransportAgent(resourceType);
        DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType);
        if (null == transportAgent || null == dataProcessor) {
            Loggers.DISTRO.warn(
                "[DISTRO-INIT] Can't find component for type {}, transportAgent: {}, dataProcessor: {}",
                resourceType, transportAgent, dataProcessor);
            return false;
        }
        for (Member each : memberManager.allMembersWithoutSelf()) {
            long startTime = System.currentTimeMillis();
            try {
                Loggers.DISTRO.info("[DISTRO-INIT] load snapshot {} from {}", resourceType,
                    each.getAddress());
                DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress());
                Loggers.DISTRO.info(
                    "[DISTRO-INIT] it took {} ms to load snapshot {} from {} and snapshot size is {}.",
                    System.currentTimeMillis() - startTime, resourceType, each.getAddress(),
                    getDistroDataLength(distroData));
                boolean result = dataProcessor.processSnapshot(distroData);
                Loggers.DISTRO
                    .info("[DISTRO-INIT] load snapshot {} from {} result: {}", resourceType,
                        each.getAddress(),
                        result);
                if (result) {
                    distroComponentHolder.findDataStorage(resourceType).finishInitial();
                    return true;
                }
            } catch (Exception e) {
                Loggers.DISTRO.error("[DISTRO-INIT] load snapshot {} from {} failed.", resourceType,
                    each.getAddress(), e);
            }
        }
        return false;
    }

    @Override
    public void registerInstance(Service service, Instance instance, String clientId) {
        Service singleton = ServiceManager.getInstance().getSingleton(service);
        if (singleton.isEphemeral()) {
            throw new NacosRuntimeException(NacosException.INVALID_PARAM,
                String.format(
                    "Current service %s is ephemeral service, can't register persistent instance.",
                    singleton.getGroupedServiceName()));
        }
        final InstanceStoreRequest request = new InstanceStoreRequest();
        request.setService(service);
        request.setInstance(instance);
        request.setClientId(clientId);
        final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
            .setData(ByteString.copyFrom(serializer.serialize(request)))
            .setOperation(DataOperation.ADD.name())
            .build();
        
        try {
            protocol.write(writeRequest);
            Loggers.RAFT.info("Client registered. service={}, clientId={}, instance={}", service,
                clientId, instance);
        } catch (Exception e) {
            throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
        }
    }

    @Override
    public Response onApply(WriteRequest request) {
        final Lock lock = readLock;
        lock.lock();
        try {
            final InstanceStoreRequest instanceRequest =
                serializer.deserialize(request.getData().toByteArray());
            final DataOperation operation = DataOperation.valueOf(request.getOperation());
            switch (operation) {
                case ADD:
                    onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                        instanceRequest.getClientId());
                    break;
                case DELETE:
                    onInstanceDeregister(instanceRequest.service, instanceRequest.getClientId());
                    break;
                case CHANGE:
                    if (instanceAndServiceExist(instanceRequest)) {
                        onInstanceRegister(instanceRequest.service, instanceRequest.instance,
                            instanceRequest.getClientId());
                    }
                    break;
                default:
                    return Response.newBuilder().setSuccess(false)
                        .setErrMsg("unsupport operation : " + operation)
                        .build();
            }
            return Response.newBuilder().setSuccess(true).build();
        } catch (Exception e) {
            Loggers.RAFT.warn("Persistent client operation failed. ", e);
            return Response.newBuilder().setSuccess(false)
                .setErrMsg("Persistent client operation failed. " + e.getMessage()).build();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Adds or updates non-aggregated data.
     *
     * @throws NacosException NacosException.
     */
    public Boolean publishConfig(ConfigForm configForm, ConfigRequestInfo configRequestInfo,
        String encryptedDataKey) throws NacosException {
        configForm
            .setNamespaceId(NamespaceUtil.processNamespaceParameter(configForm.getNamespaceId()));
        Map<String, Object> configAdvanceInfo = getConfigAdvanceInfo(configForm);
        ParamUtils.checkParam(configAdvanceInfo);
        
        configForm.setEncryptedDataKey(encryptedDataKey);
        ConfigInfo configInfo = new ConfigInfo(configForm.getDataId(), configForm.getGroup(),
            configForm.getNamespaceId(), configForm.getAppName(), configForm.getContent());
        //set old md5
        if (StringUtils.isNotBlank(configRequestInfo.getCasMd5())) {
            configInfo.setMd5(configRequestInfo.getCasMd5());
        }
        configInfo.setType(configForm.getType());
        configInfo.setEncryptedDataKey(encryptedDataKey);
        
        //beta publish
        if (StringUtils.isNotBlank(configRequestInfo.getBetaIps())) {
            configForm.setGrayName(BetaGrayRule.TYPE_BETA);
            configForm.setGrayRuleExp(configRequestInfo.getBetaIps());
            configForm.setGrayVersion(BetaGrayRule.VERSION);
            configMigrateService.persistBeta(configForm, configInfo, configRequestInfo);
            configForm.setGrayPriority(Integer.MAX_VALUE);
            configMigrateService.publishConfigGrayMigrate(BetaGrayRule.TYPE_BETA, configForm,
                configRequestInfo);
            publishConfigGray(BetaGrayRule.TYPE_BETA, configForm, configRequestInfo);
            return Boolean.TRUE;
        }
        // tag publish
        if (StringUtils.isNotBlank(configForm.getTag())) {
            configForm.setGrayName(TagGrayRule.TYPE_TAG + "_" + configForm.getTag());
            configForm.setGrayRuleExp(configForm.getTag());
            configForm.setGrayVersion(TagGrayRule.VERSION);
            configForm.setGrayPriority(Integer.MAX_VALUE - 1);
            configMigrateService.persistTagv1(configForm, configInfo, configRequestInfo);
            configMigrateService.publishConfigGrayMigrate(TagGrayRule.TYPE_TAG, configForm,
                configRequestInfo);
            publishConfigGray(TagGrayRule.TYPE_TAG, configForm, configRequestInfo);
            return Boolean.TRUE;
        }
        
        ConfigOperateResult configOperateResult;
        
        configMigrateService.publishConfigMigrate(configForm, configRequestInfo,
            configForm.getEncryptedDataKey());
        
        //formal publish
        if (StringUtils.isNotBlank(configRequestInfo.getCasMd5())) {
            configOperateResult =
                configInfoPersistService.insertOrUpdateCas(configRequestInfo.getSrcIp(),
                    configForm.getSrcUser(), configInfo, configAdvanceInfo);
            if (!configOperateResult.isSuccess()) {
                LOGGER.warn(
                    "[cas-publish-config-fail] srcIp = {}, dataId= {}, casMd5 = {}, msg = server md5 may have changed.",
                    configRequestInfo.getSrcIp(), configForm.getDataId(),
                    configRequestInfo.getCasMd5());
                throw new NacosApiException(HttpStatus.INTERNAL_SERVER_ERROR.value(),
                    ErrorCode.RESOURCE_CONFLICT,
                    "Cas publish fail, server md5 may have changed.");
            }
        } else {
            if (configRequestInfo.getUpdateForExist()) {
                configOperateResult =
                    configInfoPersistService.insertOrUpdate(configRequestInfo.getSrcIp(),
                        configForm.getSrcUser(), configInfo, configAdvanceInfo);
            } else {
                try {
                    configOperateResult =
                        configInfoPersistService.addConfigInfo(configRequestInfo.getSrcIp(),
                            configForm.getSrcUser(), configInfo, configAdvanceInfo);
                } catch (DataIntegrityViolationException ive) {
                    configOperateResult = new ConfigOperateResult(false);
                }
            }
        }
        if (!configOperateResult.isSuccess()) {
            LOGGER.warn(
                "[publish-config-failed] config already exists. dataId: {}, group: {}, namespaceId: {}",
                configForm.getDataId(), configForm.getGroup(), configForm.getNamespaceId());
            throw new ConfigAlreadyExistsException(
                String.format("config already exist, dataId: %s, group: %s, namespaceId: %s",
                    configForm.getDataId(), configForm.getGroup(), configForm.getNamespaceId()));
        }
        ConfigChangePublisher.notifyConfigChange(
            new ConfigDataChangeEvent(configForm.getDataId(), configForm.getGroup(),
                configForm.getNamespaceId(),
                configOperateResult.getLastModified()));
        if (ConfigTagUtil.isIstio(configForm.getConfigTags())) {
            ConfigChangePublisher.notifyConfigChange(
                new IstioConfigChangeEvent(configForm.getDataId(), configForm.getGroup(),
                    configForm.getNamespaceId(),
                    configOperateResult.getLastModified(), configForm.getContent(),
                    ConfigTagUtil.getIstioType(configForm.getConfigTags())));
        }
        ConfigTraceService.logPersistenceEvent(configForm.getDataId(), configForm.getGroup(),
            configForm.getNamespaceId(), configRequestInfo.getRequestIpApp(),
            configOperateResult.getLastModified(),
            InetUtils.getSelfIP(), ConfigTraceService.PERSISTENCE_EVENT,
            ConfigTraceService.PERSISTENCE_TYPE_PUB,
            configForm.getContent());
        
        return true;
    }

    @Override
    public ConfigOperateResult updateConfigInfo(final ConfigInfo configInfo, final String srcIp,
        final String srcUser,
        final Map<String, Object> configAdvanceInfo) {
        return tjt.execute(status -> {
            try {
                ConfigAllInfo oldConfigAllInfo =
                    findConfigAllInfo(configInfo.getDataId(), configInfo.getGroup(),
                        configInfo.getTenant());
                if (oldConfigAllInfo == null) {
                    if (LogUtil.FATAL_LOG.isErrorEnabled()) {
                        LogUtil.FATAL_LOG.error(
                            "expected config info[dataid:{}, group:{}, tenent:{}] but not found.",
                            configInfo.getDataId(), configInfo.getGroup(), configInfo.getTenant());
                    }
                    return new ConfigOperateResult(false);
                }
                
                String appNameTmp = oldConfigAllInfo.getAppName();
                /*
                 If the appName passed by the user is not empty, use the persistent user's appName,
                 otherwise use db; when emptying appName, you need to pass an empty string
                 */
                if (configInfo.getAppName() == null) {
                    configInfo.setAppName(appNameTmp);
                }
                updateConfigInfoAtomic(configInfo, srcIp, srcUser, configAdvanceInfo);
                String configTags = configAdvanceInfo == null ? null
                    : (String) configAdvanceInfo.get("config_tags");
                if (configTags != null) {
                    // delete all tags and then recreate
                    removeTagByIdAtomic(oldConfigAllInfo.getId());
                    addConfigTagsRelation(oldConfigAllInfo.getId(), configTags,
                        configInfo.getDataId(),
                        configInfo.getGroup(), configInfo.getTenant());
                }
                if (!ConfigPersistContext.isSkipHistory()) {
                    Timestamp now = new Timestamp(System.currentTimeMillis());
                    historyConfigInfoPersistService.insertConfigHistoryAtomic(
                        oldConfigAllInfo.getId(),
                        oldConfigAllInfo, srcIp, srcUser, now, "U", Constants.FORMAL, null,
                        ConfigExtInfoUtil.getExtInfoFromAllInfo(oldConfigAllInfo));
                }
                return getConfigInfoOperateResult(configInfo.getDataId(), configInfo.getGroup(),
                    configInfo.getTenant());
            } catch (CannotGetJdbcConnectionException e) {
                LogUtil.FATAL_LOG.error("[db-error] " + e, e);
                throw e;
            }
        });
    }

    @Override
    public Boolean update(List<ModifyRequest> sqlContext, BiConsumer<Boolean, Throwable> consumer) {
        try {
            
            // Since the SQL parameter is Object[], in order to ensure that the types of
            // array elements are not lost, the serialization here is done using the java-specific
            // serialization framework, rather than continuing with the protobuff
            
            LoggerUtils.printIfDebugEnabled(LOGGER, "modifyRequests info : {}", sqlContext);
            
            // {timestamp}-{group}-{ip:port}-{signature}
            
            final String key =
                System.currentTimeMillis() + "-" + group() + "-"
                    + memberManager.getSelf().getAddress() + "-"
                    + MD5Utils.md5Hex(sqlContext.toString(), PersistenceConstant.DEFAULT_ENCODE);
            WriteRequest request = WriteRequest.newBuilder().setGroup(group()).setKey(key)
                .setData(ByteString.copyFrom(serializer.serialize(sqlContext)))
                .putAllExtendInfo(EmbeddedStorageContextHolder.getCurrentExtendInfo())
                .setType(sqlContext.getClass().getCanonicalName()).build();
            if (Objects.isNull(consumer)) {
                Response response = this.protocol.write(request);
                if (response.getSuccess()) {
                    return true;
                }
                LOGGER.error("execute sql modify operation failed : {}", response.getErrMsg());
                return false;
            } else {
                this.protocol.writeAsync(request)
                    .whenComplete((BiConsumer<Response, Throwable>) (response, ex) -> {
                        String errMsg = Objects.isNull(ex) ? response.getErrMsg()
                            : ExceptionUtil.getCause(ex).getMessage();
                        consumer.accept(response.getSuccess(),
                            StringUtils.isBlank(errMsg) ? null : new NJdbcException(errMsg));
                    });
            }
            return true;
        } catch (TimeoutException e) {
            LOGGER.error("An timeout exception occurred during the update operation");
            throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
        } catch (Throwable e) {
            LOGGER.error("An exception occurred during the update operation : {}", e);
            throw new NacosRuntimeException(NacosException.SERVER_ERROR, e.toString());
        }
    }

    /**
     * Save config file and update md5 value in cache.
     *
     * @param dataId         dataId string value.
     * @param group          group string value.
     * @param tenant         tenant string value.
     * @param content        content string value.
     * @param md5            md5 of persist.
     * @param lastModifiedTs lastModifiedTs.
     * @param type           file type.
     * @return dumpChange success or not.
     */
    public static boolean dumpWithMd5(String dataId, String group, String tenant, String content,
        String md5,
        long lastModifiedTs, String type, String encryptedDataKey) {
        String groupKey = GroupKey2.getKey(dataId, group, tenant);
        CacheItem ci = makeSure(groupKey, encryptedDataKey);
        ci.setType(type);
        final int lockResult = tryWriteLock(groupKey);
        
        if (lockResult < 0) {
            DUMP_LOG.warn("[dump-error] write lock failed. {}", groupKey);
            return false;
        }
        
        try {
            
            //check timestamp
            boolean lastModifiedOutDated =
                lastModifiedTs < ConfigCacheService.getLastModifiedTs(groupKey);
            if (lastModifiedOutDated) {
                DUMP_LOG.warn("[dump-ignore] timestamp is outdated,groupKey={}", groupKey);
                return true;
            }
            
            boolean newLastModified =
                lastModifiedTs > ConfigCacheService.getLastModifiedTs(groupKey);
            
            if (md5 == null) {
                md5 = MD5Utils.md5Hex(content, PERSIST_ENCODE);
            }
            
            //check md5 & update local disk cache.
            String localContentMd5 = ConfigCacheService.getContentMd5(groupKey);
            boolean md5Changed = !md5.equals(localContentMd5);
            if (md5Changed) {
                DUMP_LOG.info(
                    "[dump] md5 changed, save to disk cache ,groupKey={}, newMd5={},oldMd5={}",
                    groupKey, md5,
                    localContentMd5);
                ConfigDiskServiceFactory.getInstance().saveToDisk(dataId, group, tenant, content);
            } else {
                DUMP_LOG.warn(
                    "[dump-ignore] ignore to save to disk cache. md5 consistent,groupKey={}, md5={}",
                    groupKey, md5);
            }
            
            //check  md5 and timestamp & update local jvm cache.
            if (md5Changed) {
                DUMP_LOG.info(
                    "[dump] md5 changed, update md5 and timestamp in jvm cache ,groupKey={}, newMd5={},oldMd5={},lastModifiedTs={}",
                    groupKey, md5, localContentMd5, lastModifiedTs);
                updateMd5(groupKey, md5, content, lastModifiedTs, encryptedDataKey);
            } else if (newLastModified) {
                DUMP_LOG.info(
                    "[dump] md5 consistent ,timestamp changed, update timestamp only in jvm cache ,groupKey={},lastModifiedTs={}",
                    groupKey, lastModifiedTs);
                updateTimeStamp(groupKey, lastModifiedTs, encryptedDataKey);
            } else {
                DUMP_LOG.warn(
                    "[dump-ignore] ignore to save to jvm cache. md5 consistent and no new timestamp changed.groupKey={}",
                    groupKey);
            }
            
            return true;
        } catch (IOException ioe) {
            DUMP_LOG.error("[dump-exception] save disk error. " + groupKey + ", " + ioe);
            if (ioe.getMessage() != null) {
                String errMsg = ioe.getMessage();
                if (errMsg.contains(NO_SPACE_CN) || errMsg.contains(NO_SPACE_EN)
                    || errMsg.contains(DISK_QUOTA_CN)
                    || errMsg.contains(DISK_QUOTA_EN)) {
                    // Protect from disk full.
                    FATAL_LOG.error("Local Disk Full,Exit", ioe);
                    EnvUtil.systemExit();
                }
            }
            return false;
        } finally {
            releaseWriteLock(groupKey);
        }
        
    }

    @Override
    @NamespaceValidation
    @TpsControl(pointName = "ConfigListen")
    @Secured(action = ActionTypes.READ, signType = SignType.CONFIG)
    @ExtractorManager.Extractor(rpcExtractor = ConfigBatchListenRequestParamExtractor.class)
    public ConfigChangeBatchListenResponse handle(
        ConfigBatchListenRequest configChangeListenRequest, RequestMeta meta)
        throws NacosException {
        String connectionId = StringPool.get(meta.getConnectionId());
        String tag = configChangeListenRequest.getHeader(Constants.VIPSERVER_TAG);
        ParamUtils.checkParam(tag);
        ConfigChangeBatchListenResponse configChangeBatchListenResponse =
            new ConfigChangeBatchListenResponse();
        for (ConfigBatchListenRequest.ConfigListenContext listenContext : configChangeListenRequest
            .getConfigListenContexts()) {
            boolean isNeedTransferNamespace =
                NamespaceUtil.isNeedTransferNamespace(listenContext.getTenant());
            String namespaceId = NamespaceUtil.processNamespaceParameter(listenContext.getTenant());
            String groupKey =
                GroupKey2.getKey(listenContext.getDataId(), listenContext.getGroup(), namespaceId);
            groupKey = StringPool.get(groupKey);
            
            String md5 = StringPool.get(listenContext.getMd5());
            
            if (configChangeListenRequest.isListen()) {
                configChangeListenContext.addListen(groupKey, md5, connectionId,
                    isNeedTransferNamespace);
                boolean isUptoDate =
                    ConfigCacheService.isUptodate(groupKey, md5, meta.getClientIp(), tag,
                        meta.getAppLabels());
                if (!isUptoDate) {
                    configChangeBatchListenResponse.addChangeConfig(listenContext.getDataId(),
                        listenContext.getGroup(),
                        listenContext.getTenant());
                }
            } else {
                configChangeListenContext.removeListen(groupKey, connectionId);
            }
        }
        
        return configChangeBatchListenResponse;
        
    }

        ConfigChangeNotifyResponse handleConfigChangeNotifyRequest(
            ConfigChangeNotifyRequest configChangeNotifyRequest,
            String clientName) {
            LOGGER.info("[{}] [server-push] config changed. dataId={}, group={},tenant={}",
                clientName,
                configChangeNotifyRequest.getDataId(), configChangeNotifyRequest.getGroup(),
                configChangeNotifyRequest.getTenant());
            String groupKey = GroupKey.getKeyTenant(configChangeNotifyRequest.getDataId(),
                configChangeNotifyRequest.getGroup(), configChangeNotifyRequest.getTenant());
            
            CacheData cacheData = cacheMap.get().get(groupKey);
            if (cacheData != null) {
                synchronized (cacheData) {
                    cacheData.getReceiveNotifyChanged().set(true);
                    cacheData.setConsistentWithServer(false);
                    notifyListenConfig();
                }
                
            }
            return new ConfigChangeNotifyResponse();
        }

    private String getConfigInner(String tenant, String dataId, String group, long timeoutMs)
        throws NacosException {
        group = blank2defaultGroup(group);
        ParamUtils.checkKeyParam(dataId, group);
        ConfigResponse cr = new ConfigResponse();
        
        cr.setDataId(dataId);
        cr.setTenant(tenant);
        cr.setGroup(group);
        
        // We first try to use local failover content if exists.
        // A config content for failover is not created by client program automatically,
        // but is maintained by user.
        // This is designed for certain scenario like client emergency reboot,
        // changing config needed in the same time, while nacos server is down.
        String content =
            LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);
        if (content != null) {
            LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}",
                worker.getAgentName(),
                dataId, group, tenant);
            cr.setContent(content);
            String encryptedDataKey =
                LocalEncryptedDataKeyProcessor.getEncryptDataKeyFailover(worker.getAgentName(),
                    dataId, group, tenant);
            cr.setEncryptedDataKey(encryptedDataKey);
            configFilterChainManager.doFilter(null, cr);
            content = cr.getContent();
            return content;
        }
        
        try {
            ConfigResponse response =
                worker.getServerConfig(dataId, group, tenant, timeoutMs, false);
            cr.setContent(response.getContent());
            cr.setEncryptedDataKey(response.getEncryptedDataKey());
            configFilterChainManager.doFilter(null, cr);
            content = cr.getContent();
            
            return content;
        } catch (NacosException ioe) {
            if (NacosException.NO_RIGHT == ioe.getErrCode()) {
                throw ioe;
            }
            LOGGER.warn(
                "[{}] [get-config] get from server error, dataId={}, group={}, tenant={}, msg={}",
                worker.getAgentName(), dataId, group, tenant, ioe.toString());
        }
        
        content =
            LocalConfigInfoProcessor.getSnapshot(worker.getAgentName(), dataId, group, tenant);
        if (content != null) {
            LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}",
                worker.getAgentName(),
                dataId, group, tenant);
        }
        cr.setContent(content);
        String encryptedDataKey =
            LocalEncryptedDataKeyProcessor.getEncryptDataKeySnapshot(worker.getAgentName(),
                dataId, group, tenant);
        cr.setEncryptedDataKey(encryptedDataKey);
        configFilterChainManager.doFilter(null, cr);
        content = cr.getContent();
        return content;
    }

    /**
     * Checks the complete member view and irreversibly enables JRaft authentication when every
     * member reports support. In-memory enforcement is enabled immediately, while state-file
     * persistence is retried independently until it succeeds.
     */
    @Scheduled(fixedRate = 3000)
    public synchronized void doCheck() {
        if (enforced.get()) {
            persistStateIfNecessary();
            return;
        }
        Collection<Member> members = serverMemberManager.allMembers();
        if (members == null || members.isEmpty()) {
            return;
        }
        for (Member each : members) {
            if (!Boolean.TRUE.equals(
                each.getExtendVal(MemberMetaDataConstants.SUPPORT_JRAFT_AUTH))) {
                return;
            }
        }
        enforced.set(true);
        Loggers.RAFT.info(
            "All Nacos servers support JRaft authentication; enforcement is enabled");
        persistStateIfNecessary();
    }

    /**
     * Create new mcp server with resource specification.
     *
     * @param namespaceId namespace id of mcp server
     * @param serverSpecification mcp server specification
     * @param toolSpecification mcp server tool specification
     * @param resourceSpecification mcp server resource specification
     * @param endpointSpecification mcp server endpoint specification
     * @return mcp server id
     * @throws NacosException any exception during handling
     */
    public String createMcpServer(String namespaceId, McpServerBasicInfo serverSpecification,
        McpToolSpecification toolSpecification, McpResourceSpecification resourceSpecification,
        McpEndpointSpec endpointSpecification) throws NacosException {
        
        String existId =
            resolveMcpServerId(namespaceId, serverSpecification.getName(), StringUtils.EMPTY);
        if (StringUtils.isNotEmpty(existId)) {
            throw new NacosApiException(NacosApiException.CONFLICT, ErrorCode.RESOURCE_CONFLICT,
                String.format("mcp server `%s` has existed, please update it rather than create.",
                    serverSpecification.getName()));
        }
        
        ServerVersionDetail versionDetail = serverSpecification.getVersionDetail();
        if (null == versionDetail && StringUtils.isNotBlank(serverSpecification.getVersion())) {
            versionDetail = new ServerVersionDetail();
            versionDetail.setVersion(serverSpecification.getVersion());
            serverSpecification.setVersionDetail(versionDetail);
        }
        if (Objects.isNull(versionDetail) || StringUtils.isEmpty(versionDetail.getVersion())) {
            throw new NacosApiException(NacosApiException.INVALID_PARAM,
                ErrorCode.PARAMETER_VALIDATE_ERROR,
                "Version must be specified in parameter `serverSpecification`");
        }
        String id;
        String customMcpId = serverSpecification.getId();
        
        if (StringUtils.isEmpty(customMcpId)) {
            id = UUID.randomUUID().toString();
        } else {
            if (!StringUtils.isUuidString(customMcpId)) {
                throw new NacosApiException(NacosApiException.INVALID_PARAM,
                    ErrorCode.PARAMETER_VALIDATE_ERROR,
                    "parameter `serverSpecification.id` is not match uuid pattern,  must obey uuid pattern");
            }
            if (mcpServerIndex.getMcpServerById(serverSpecification.getId()) != null) {
                throw new NacosApiException(NacosApiException.INVALID_PARAM,
                    ErrorCode.PARAMETER_VALIDATE_ERROR,
                    "parameter `serverSpecification.id` conflict with exist mcp server id");
            }
            
            id = customMcpId;
        }
        
        serverSpecification.setId(id);
        ZonedDateTime currentTime = ZonedDateTime.now(ZoneOffset.UTC);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern(Constants.RELEASE_DATE_FORMAT);
        String formattedCurrentTime = currentTime.format(formatter);
        versionDetail.setRelease_date(formattedCurrentTime);
        
        McpServerStorageInfo newSpecification = new McpServerStorageInfo();
        BeanUtils.copyProperties(serverSpecification, newSpecification);
        injectMcpDescriptionsAndEndpoint(namespaceId, serverSpecification.getId(), newSpecification,
            toolSpecification, resourceSpecification, endpointSpecification, false, null);
        
        McpServerVersionInfo versionInfo =
            buildServerVersionInfo(newSpecification, id, versionDetail);
        
        ConfigRequestInfo configRequestInfo = new ConfigRequestInfo();
        configRequestInfo.setUpdateForExist(Boolean.FALSE);
        
        ConfigFormV3 mcpServerVersionForm = buildMcpServerVersionForm(namespaceId, versionInfo);
        configOperationService.publishConfig(mcpServerVersionForm, configRequestInfo, null);
        
        ConfigForm configForm =
            buildMcpConfigForm(namespaceId, id, versionDetail.getVersion(), newSpecification);
        long startOperationTime = System.currentTimeMillis();
        configOperationService.publishConfig(configForm, configRequestInfo, null);
        syncEffectService.toSync(configForm, startOperationTime);
        
        // Delete the relevant cache after a successful database operation
        invalidateCacheAfterDbOperation(namespaceId, serverSpecification.getName(), id);
        AiResourceTraceService.logSuccess("mcp", serverSpecification.getName(),
            versionDetail.getVersion(), AiResourceTraceService.OP_CREATE_DRAFT,
            VisibilityHelper.resolveCurrentIdentity(), VisibilityHelper.resolveClientIp());
        
        return id;
    }

    // ========== Client APIs ==========
    
    @Override
    public PromptVersionInfo queryPrompt(String namespaceId, String promptKey, String version,
        String label)
        throws NacosException {
        AiResource meta = requireReadableMeta(namespaceId, promptKey);
        
        PromptVersionInfoPojo info = requireVersionInfo(meta);
        String resolved = resolveClientVersion(info, version, label);
        if (StringUtils.isBlank(resolved)) {
            throw new NacosApiException(NacosException.NOT_FOUND, ErrorCode.RESOURCE_NOT_FOUND,
                "Prompt version not found: " + promptKey);
        }
        
        // Verify version is online
        AiResourceVersion versionRow = resourceManager.findVersion(namespaceId, promptKey,
            RESOURCE_TYPE_PROMPT, resolved);
        if (versionRow == null || !VERSION_STATUS_ONLINE.equalsIgnoreCase(versionRow.getStatus())) {
            throw new NacosApiException(NacosException.NOT_FOUND, ErrorCode.RESOURCE_NOT_FOUND,
                "Prompt version not online: " + promptKey + "@" + resolved);
        }
        
        PromptVersionInfo result = loadPromptFromStorage(namespaceId, promptKey, resolved);
        result.setSrcUser(versionRow.getAuthor());
        result.setCommitMsg(versionRow.getDesc());
        return result;
    }