@Override
    public boolean process(NacosTask task) {
        MergeDataTask mergeTask = (MergeDataTask) task;
        final String dataId = mergeTask.dataId;
        final String group = mergeTask.groupId;
        final String tenant = mergeTask.tenant;
        final String tag = mergeTask.tag;
        final String clientIp = mergeTask.getClientIp();
        try {
            List<ConfigInfoAggr> datumList = new ArrayList<>();
            int rowCount = configInfoAggrPersistService.aggrConfigInfoCount(dataId, group, tenant);
            int pageCount = (int) Math.ceil(rowCount * 1.0 / PAGE_SIZE);
            for (int pageNo = 1; pageNo <= pageCount; pageNo++) {
                Page<ConfigInfoAggr> page = configInfoAggrPersistService.findConfigInfoAggrByPage(dataId, group, tenant,
                        pageNo, PAGE_SIZE);
                if (page != null) {
                    datumList.addAll(page.getPageItems());
                    LOGGER.info("[merge-query] {}, {}, size/total={}/{}", dataId, group, datumList.size(), rowCount);
                }
            }
            
            final Timestamp time = TimeUtils.getCurrentTime();
            if (datumList.size() > 0) {
                // merge
                ConfigInfo cf = merge(dataId, group, tenant, datumList);
                
                configInfoPersistService.insertOrUpdate(null, null, cf, null);
                
                LOGGER.info("[merge-ok] {}, {}, size={}, length={}, md5={}, content={}", dataId, group,
                        datumList.size(), cf.getContent().length(), cf.getMd5(),
                        ContentUtils.truncateContent(cf.getContent()));
                
                ConfigTraceService.logPersistenceEvent(dataId, group, tenant, null, time.getTime(),
                        InetUtils.getSelfIP(), ConfigTraceService.PERSISTENCE_EVENT,
                        ConfigTraceService.PERSISTENCE_TYPE_MERGE, cf.getContent());
            } else {
                String eventType;
                
                // remove
                if (StringUtils.isBlank(tag)) {
                    eventType = ConfigTraceService.PERSISTENCE_EVENT;
                    
                    configInfoPersistService.removeConfigInfo(dataId, group, tenant, clientIp, null);
                } else {
                    eventType = ConfigTraceService.PERSISTENCE_EVENT_TAG + "-" + tag;
                    
                    configInfoTagPersistService.removeConfigInfoTag(dataId, group, tenant, tag, clientIp, null);
                }
                
                LOGGER.warn(
                        "[merge-delete] delete config info because no datum. dataId=" + dataId + ", groupId=" + group);
                
                ConfigTraceService.logPersistenceEvent(dataId, group, tenant, null, time.getTime(),
                        InetUtils.getSelfIP(), eventType, ConfigTraceService.PERSISTENCE_TYPE_REMOVE, null);
            }
            NotifyCenter.publishEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));
            
        } catch (Exception e) {
            mergeService.addMergeTask(dataId, group, tenant, mergeTask.getClientIp());
            LOGGER.info("[merge-error] " + dataId + ", " + group + ", " + e.toString(), e);
        }
        
        return true;
    }
@Test
    void testMergerExistAggrConfig() throws InterruptedException {
        String dataId = "dataId12345";
        String group = "group123";
        String tenant = "tenant1234";
        when(configInfoAggrPersistService.aggrConfigInfoCount(eq(dataId), eq(group), eq(tenant))).thenReturn(2);
        Page<ConfigInfoAggr> datumPage = new Page<>();
        ConfigInfoAggr configInfoAggr1 = new ConfigInfoAggr();
        configInfoAggr1.setContent("12344");
        ConfigInfoAggr configInfoAggr2 = new ConfigInfoAggr();
        configInfoAggr2.setContent("12345666");
        datumPage.getPageItems().add(configInfoAggr1);
        datumPage.getPageItems().add(configInfoAggr2);
        
        when(configInfoAggrPersistService.findConfigInfoAggrByPage(eq(dataId), eq(group), eq(tenant), anyInt(), anyInt())).thenReturn(
                datumPage);
        
        when(configInfoPersistService.insertOrUpdate(eq(null), eq(null), any(ConfigInfo.class), eq(null))).thenReturn(
                new ConfigOperateResult());
        
        AtomicReference<ConfigDataChangeEvent> reference = new AtomicReference<>();
        NotifyCenter.registerSubscriber(new Subscriber() {
            
            @Override
            public void onEvent(Event event) {
                ConfigDataChangeEvent event1 = (ConfigDataChangeEvent) event;
                if (event1.dataId.equals(dataId) && event1.group.equals(group) && tenant.equals(event1.tenant)) {
                    reference.set((ConfigDataChangeEvent) event);
                }
            }
            
            @Override
            public Class<? extends Event> subscribeType() {
                return ConfigDataChangeEvent.class;
            }
        });
        
        MergeDataTask mergeDataTask = new MergeDataTask(dataId, group, tenant, "127.0.0.1");
        mergeTaskProcessor.process(mergeDataTask);
        
        Mockito.verify(configInfoPersistService, times(1)).insertOrUpdate(eq(null), eq(null), any(ConfigInfo.class), eq(null));
        Thread.sleep(1000L);
        assertTrue(reference.get() != null);
        
    }