@Override
    @Transactional(rollbackFor = Throwable.class, isolation = Isolation.REPEATABLE_READ, propagation = Propagation.REQUIRES_NEW)
    public Boolean startTagSwitch(String groupId, String clusterTag) {
        LOGGER.info("start to switch cluster tag for group={}, target tag={}", groupId, clusterTag);

        InlongGroupInfo groupInfo = this.get(groupId);

        // check if the group mode is data sync mode
        if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
            String errMSg = String.format("no need to switch sync mode group = {}", groupId);
            LOGGER.error(errMSg);
            throw new BusinessException(errMSg);
        }

        // check if the group is under switching
        List<InlongGroupExtInfo> groupExt = groupInfo.getExtList();
        Set<String> keys = groupExt.stream()
                .map(InlongGroupExtInfo::getKeyName)
                .collect(Collectors.toSet());

        if (keys.contains(BACKUP_CLUSTER_TAG) || keys.contains(BACKUP_MQ_RESOURCE)) {
            String errMsg = String.format("switch failed, current group is under switching, group=[%s]", groupId);
            LOGGER.error(errMsg);
            throw new BusinessException(errMsg);
        }

        // check if the cluster tag is under current tenant
        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
        if (groupEntity == null) {
            LOGGER.error("inlong group not found by groupId={}", groupId);
            throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND);
        }

        TenantClusterTagEntity tenantClusterTag =
                tenantClusterTagMapper.selectByUniqueKey(clusterTag, groupEntity.getTenant());
        if (tenantClusterTag == null) {
            LOGGER.error("tenant cluster not found for tenant={}, clusterTag={}", groupEntity.getTenant(), clusterTag);
            throw new BusinessException(ErrorCodeEnum.TENANT_CLUSTER_TAG_NOT_FOUND);
        }

        // check if all sink related sort cluster has the target cluster tag
        List<StreamSink> sinks = streamSinkService.listSink(groupEntity.getInlongGroupId(), null);
        for (StreamSink sink : sinks) {
            String clusterName = sink.getInlongClusterName();
            List<InlongClusterEntity> clusterEntity =
                    clusterEntityMapper.selectByKey(clusterTag, clusterName,
                            SinkType.relatedSortClusterType(sink.getSinkType()));
            if (CollectionUtils.isEmpty(clusterEntity) || clusterEntity.size() != 1) {
                String errMsg = String.format("find no cluster or multiple cluster with clusterName=[%s]", clusterName);
                LOGGER.error(errMsg);
                throw new BusinessException(ErrorCodeEnum.CLUSTER_NOT_FOUND, errMsg);
            }

        }

        // config cluster tag and backup_cluster_tag
        UserInfo userInfo = LoginUserUtils.getLoginUser();
        InlongGroupRequest request = groupInfo.genRequest();
        String oldClusterTag = request.getInlongClusterTag();
        request.setInlongClusterTag(clusterTag);
        request.getExtList().add(new InlongGroupExtInfo(null, groupId, BACKUP_CLUSTER_TAG, oldClusterTag));
        request.getExtList().add(new InlongGroupExtInfo(null, groupId, BACKUP_MQ_RESOURCE, request.getMqResource()));
        request.getExtList().add(new InlongGroupExtInfo(null, groupId, CLUSTER_SWITCH_TIME,
                LocalDateTime.now().toString()));
        this.update(request, userInfo.getName());

        // trigger group workflow to rebuild configs
        this.triggerWorkFlow(groupInfo, userInfo);
        LOGGER.info("success to switch cluster tag for group={}, target tag={}", groupId, clusterTag);
        return true;
    }
Add constraints to the start and end time of the offline synchronization group.
1. startTime must >= current time
2. endTime must >= startTime
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java