public static void main(String[] args) throws AdminException {
        AdminDistributedSystem adminDs = AdminDistributedSystemFactory.getDistributedSystem(
                AdminDistributedSystemFactory.defineDistributedSystem()
        );
        adminDs.connect();

        //connect to different clusters. test data                             todo: insert cli usage
        Pool poolForCluster1 = PoolManager.createFactory().addLocator("localhost", 10330).create("cluster1");
        Pool poolForCluster2 = PoolManager.createFactory().addLocator("localhost", 10332).create("cluster2");
        //Pool poolForServer =  PoolManager.createFactory().addServer("localhost", 40405).create("node");
        List<Pool> poolList = new ArrayList<Pool>();
        poolList.add(poolForCluster1);
        poolList.add(poolForCluster2);
        //poolList.add(poolForServer);

        //todo: insert checking that each cluster contains region and one's type is equal (Partitioned, Replicated)

        boolean partitioned = true;        //todo: insert CLI usage  + throw exception if real region has another type


        List<ServerLocation> serverFromPool = new ArrayList<ServerLocation>();
        List<Pool> emptyPools = new ArrayList<Pool>(); //contains pool with no available servers
        for (Pool pool : poolList) {
            List<ServerLocation> allServers = null;
            if (!pool.getLocators().isEmpty())
                allServers = ((AutoConnectionSourceImpl) ((PoolImpl) pool).getConnectionSource()).findAllServers();
            else if (!pool.getServers().isEmpty())
                allServers = Arrays.asList((((PoolImpl) pool).getConnectionSource()).findServer(Collections.emptySet()));

            if (allServers != null)
                serverFromPool.addAll(allServers);
            else {
                logger.info("not found servers on locator {}", pool);
                emptyPools.add(pool);
            }
        }
        poolList.removeAll(emptyPools);

        if (serverFromPool.size() == 0) {
            logger.info("no servers available");
            return;
        }

        for (ServerLocation server : serverFromPool) {
            System.out.println("host: " + server.getHostName());
            System.out.println("port: " + server.getPort());
            System.out.println("----------------------");
        }

        //source for comparison //todo: if this node doesn't contain region! it's problem
        Pool sourcePool;
        if (!partitioned) {
            int randomServerLocation = new Random().nextInt(serverFromPool.size());
            sourcePool = PoolManager.createFactory()
                    .addServer(serverFromPool.get(randomServerLocation).getHostName(),
                            serverFromPool.get(randomServerLocation).getPort()).create("target");
        } else {
            sourcePool = poolList.get(0);
            poolList.remove(0);
        }


        FunctionService.registerFunction(new RegionInfoFunction());
        ResultCollector regionInfoResult = FunctionService.onServers(sourcePool)
                .withArgs(REGION_NAME)
                .execute(new RegionInfoFunction());

        Map regionInfo = (HashMap) ((ArrayList) regionInfoResult.getResult()).get(0);
        System.out.println("region info: " + regionInfo);

        int totalNumBuckets = (Integer) regionInfo.get("totalNumBuckets");
        //logger.debug("total keys' batch counts is ", totalNumBuckets);
        System.out.println("total keys' batch counts is " + totalNumBuckets);
        KeyExtractor keyExtractor = new KeyExtractor(REGION_NAME, sourcePool, partitioned, totalNumBuckets);


        Map<String, Map<String, Set>> clusterDifference = new HashMap<String, Map<String, Set>>();          //key: memeberId list: absent keys, diff values

        //List<ResultCollector> taskResults = new ArrayList<ResultCollector>();
        List<PoolResult> taskResults = new ArrayList<PoolResult>();
        //List<Future<ResultCollector>> collectTasks = new ArrayList<Future<ResultCollector>>(poolList.size());
        List<Future<PoolResult>> collectTasks = new ArrayList<Future<PoolResult>>(poolList.size());
        ExecutorService executorService = Executors.newFixedThreadPool(poolList.size());
        while (keyExtractor.hasKeys()) {
            Set keys = keyExtractor.getNextKeysBatch();
            System.out.println("keys to check: " + keys);
            for (Pool nextPool : poolList)
                collectTasks.add(executorService.submit(new CollectorTask(keys, nextPool, REGION_NAME)));
            System.out.println("active tasks: " + collectTasks.size());
            try {
                //for (Future<ResultCollector> futureTask : collectTasks) {
                for (Future<PoolResult> futureTask : collectTasks) {
                    taskResults.add(futureTask.get());
                }
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            } catch (ExecutionException ee) {
                ee.printStackTrace();
            }
            collectTasks.clear();

            System.out.println("compare contents..");
            //getting original contents
            Map originalData = new HashMap();
            /*           for (ResultCollector taskResultFromPool : taskResults) {
                List memberResult = (ArrayList) taskResultFromPool.getResult();
                if (!partitioned)
                    for (Object result : memberResult) {                        //if cluster has several nodes with replicated region... todo: adopt for partitioned
                        if (regionInfo.get("id").equals(((HashMap) result).get("memberId"))) {
                            originalData = (HashMap) ((HashMap) result).get("map");
                            break;
                        }
                    }
                else {
                    FutureTask<ResultCollector> ft = new FutureTask<ResultCollector>(new CollectorTask(keys, sourcePool, REGION_NAME));
                    try {
                        ResultCollector rc = ft.get();
                    } catch (Exception e) {
                        throw new RuntimeException("error getting key-hashcode from pool: " + sourcePool, e);
                    }
                }
            }*/
            //FutureTask<ResultCollector> ft = new FutureTask<ResultCollector>(new CollectorTask(keys, sourcePool, REGION_NAME));
            FutureTask<PoolResult> ft = new FutureTask<PoolResult>(new CollectorTask(keys, sourcePool, REGION_NAME));
            ft.run();
            try {
                //ResultCollector rc = ft.get();
                PoolResult rc = ft.get();
                List poolResult = (List) rc.getResultCollector().getResult();
                for (Object singleResult : poolResult) {
                    originalData.putAll((Map) ((HashMap) singleResult).get("map"));
                }
            } catch (Exception e) {
                throw new RuntimeException("error getting key-hash from pool: " + sourcePool, e);
            }

            /*    if (true)
            return;

        //todo: aggregate members' data from one cluster
        if (partitioned)
            aggregateSingleClusterMemberData(taskResults);*/

            System.out.println("original data is: " + originalData);
            //for (ResultCollector taskResultFromPool : taskResults) {
            for (PoolResult taskResultFromPool : taskResults) {
                List poolResult = (ArrayList) taskResultFromPool.getResultCollector().getResult();
                if (!partitioned) {
                    for (Object resultFromMember : poolResult) {
                        Map result = (HashMap) resultFromMember;
                        if (regionInfo.get("id").equals(result.get("memberId")))                   //for replicated region
                            continue;
                      /*  System.out.println("member id: " + result.get("memberId"));
                        System.out.println("absent keys is " + result.get("absentKeys"));
                        System.out.println("it's data is: " + result.get("map"));*/
                        String memberId = (String) result.get("memberId");
                        Map memberData = (HashMap) result.get("map");
                        Map memberDataCopy = new HashMap();
                        memberDataCopy.putAll(memberData);
                        memberDataCopy.keySet().retainAll(originalData.keySet());
                        Map<String, Set> aggregationInfo = new HashMap<String, Set>();      //contains aggregation info for a member
                        //keys witch values are different
                        Set keysForDiffValues = new HashSet();
                        Set absentKeys = new HashSet();
                        for (Object regionKey : memberDataCopy.keySet())
                            if (!memberDataCopy.get(regionKey).equals(originalData.get(regionKey)))      //we compare not original values, but it's hash code
                                keysForDiffValues.add(regionKey);

                        aggregationInfo.put("diffValues", keysForDiffValues);
                        //absent keys
                        memberData.keySet().removeAll(memberDataCopy.keySet());

                        absentKeys.addAll(memberData.keySet());
                        absentKeys.addAll((HashSet) result.get("absentKeys"));
                        aggregationInfo.put("absentKeys", absentKeys);

                       /* if (!clusterDifference.containsKey(memberId)) {
                            clusterDifference.put(memberId, aggregationInfo);
                        } else {
                            Map<String, Set> difference = clusterDifference.get(memberId);
                            difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys"));
                            difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                            clusterDifference.put(memberId, difference);
                        }*/
                         if (!clusterDifference.containsKey(memberId)) {
                            clusterDifference.put(memberId, aggregationInfo);
                        } else {
                            Map<String, Set> difference = clusterDifference.get(memberId);
                            difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys"));
                            difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                            clusterDifference.put(memberId, difference);
                        }
                    }
                } else {
                    Map targetMap = new HashMap();
                    Set absentPoolKeys = new HashSet();     //aggregate absent keys from pool members
                    for (Object memberResult : poolResult) {
                        targetMap.putAll((Map) ((HashMap) memberResult).get("map"));
                        absentPoolKeys.addAll((Set) (((HashMap)memberResult).get("absentKeys")));
                    }
                    Map memberDataCopy = new HashMap();
                    memberDataCopy.putAll(targetMap);
                    memberDataCopy.keySet().retainAll(originalData.keySet());
                    Map<String, Set> aggregationInfo = new HashMap<String, Set>();      //contains aggregation info for a member
                    //keys witch values are different
                    Set keysForDiffValues = new HashSet();
                    Set absentKeys = new HashSet();
                    for (Object regionKey : memberDataCopy.keySet())
                        if (!memberDataCopy.get(regionKey).equals(originalData.get(regionKey)))      //we compare not original values, but it's hash code
                            keysForDiffValues.add(regionKey);

                    aggregationInfo.put("diffValues", keysForDiffValues);
                    //absent keys
                    targetMap.keySet().removeAll(memberDataCopy.keySet());

                    absentKeys.addAll(targetMap.keySet());
                    absentKeys.addAll(absentPoolKeys);
                    //aggregationInfo.put("absentKeys", absentKeys);

                    if (!clusterDifference.containsKey(taskResultFromPool.getPool().toString())) {
                        clusterDifference.put(taskResultFromPool.getPool().toString(), aggregationInfo);
                    } else {
                        Map<String, Set> difference = clusterDifference.get(taskResultFromPool.getPool().toString());
                      //  difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys"));
                        difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                        clusterDifference.put(taskResultFromPool.getPool().toString(), difference);
                    }
                }
            }

            taskResults.clear();
        }

        System.out.println("____________________________");
        System.out.println("difference: ");
        System.out.println(clusterDifference);
        executorService.shutdown();
        adminDs.disconnect();
    }
public static void main(String[] args) throws AdminException {
        AdminDistributedSystem adminDs = AdminDistributedSystemFactory.getDistributedSystem(
                AdminDistributedSystemFactory.defineDistributedSystem()
        );
        adminDs.connect();

        
        Pool poolForCluster1 = PoolManager.createFactory().addLocator("localhost", 10330).create("cluster1");
        Pool poolForCluster2 = PoolManager.createFactory().addLocator("localhost", 10332).create("cluster2");
        
        List<Pool> poolList = new ArrayList<Pool>();
        poolList.add(poolForCluster1);
        poolList.add(poolForCluster2);
        

        

        boolean partitioned = true;        


        List<ServerLocation> serverFromPool = new ArrayList<ServerLocation>();
        List<Pool> emptyPools = new ArrayList<Pool>(); 
        for (Pool pool : poolList) {
            List<ServerLocation> allServers = null;
            if (!pool.getLocators().isEmpty())
                allServers = ((AutoConnectionSourceImpl) ((PoolImpl) pool).getConnectionSource()).findAllServers();
            else if (!pool.getServers().isEmpty())
                allServers = Arrays.asList((((PoolImpl) pool).getConnectionSource()).findServer(Collections.emptySet()));

            if (allServers != null)
                serverFromPool.addAll(allServers);
            else {
                logger.info("not found servers on locator {}", pool);
                emptyPools.add(pool);
            }
        }
        poolList.removeAll(emptyPools);

        if (serverFromPool.size() == 0) {
            logger.info("no servers available");
            return;
        }

        for (ServerLocation server : serverFromPool) {
            System.out.println("host: " + server.getHostName());
            System.out.println("port: " + server.getPort());
            System.out.println("----------------------");
        }

        
        Pool sourcePool;
        if (!partitioned) {
            int randomServerLocation = new Random().nextInt(serverFromPool.size());
            sourcePool = PoolManager.createFactory()
                    .addServer(serverFromPool.get(randomServerLocation).getHostName(),
                            serverFromPool.get(randomServerLocation).getPort()).create("target");
        } else {
            sourcePool = poolList.get(0);
            poolList.remove(0);
        }


        FunctionService.registerFunction(new RegionInfoFunction());
        ResultCollector regionInfoResult = FunctionService.onServers(sourcePool)
                .withArgs(REGION_NAME)
                .execute(new RegionInfoFunction());

        Map regionInfo = (HashMap) ((ArrayList) regionInfoResult.getResult()).get(0);
        System.out.println("region info: " + regionInfo);

        int totalNumBuckets = (Integer) regionInfo.get("totalNumBuckets");
        
        System.out.println("total keys' batch counts is " + totalNumBuckets);
        KeyExtractor keyExtractor = new KeyExtractor(REGION_NAME, sourcePool, partitioned, totalNumBuckets);


        Map<String, Map<String, Set>> clusterDifference = new HashMap<String, Map<String, Set>>();          

        
        List<PoolResult> taskResults = new ArrayList<PoolResult>();
        
        List<Future<PoolResult>> collectTasks = new ArrayList<Future<PoolResult>>(poolList.size());
        ExecutorService executorService = Executors.newFixedThreadPool(poolList.size());
        while (keyExtractor.hasKeys()) {
            Set keys = keyExtractor.getNextKeysBatch();
            System.out.println("keys to check: " + keys);
            for (Pool nextPool : poolList)
                collectTasks.add(executorService.submit(new CollectorTask(keys, nextPool, REGION_NAME)));
            System.out.println("active tasks: " + collectTasks.size());
            try {
                
                for (Future<PoolResult> futureTask : collectTasks) {
                    taskResults.add(futureTask.get());
                }
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            } catch (ExecutionException ee) {
                ee.printStackTrace();
            }
            collectTasks.clear();

            System.out.println("compare contents..");
            
            Map originalData = new HashMap();
            
            
            FutureTask<PoolResult> ft = new FutureTask<PoolResult>(new CollectorTask(keys, sourcePool, REGION_NAME));
            ft.run();
            try {
                
                PoolResult rc = ft.get();
                List poolResult = (List) rc.getResultCollector().getResult();
                for (Object singleResult : poolResult) {
                    originalData.putAll((Map) ((HashMap) singleResult).get("map"));
                }
            } catch (Exception e) {
                throw new RuntimeException("error getting key-hash from pool: " + sourcePool, e);
            }

            

            System.out.println("original data is: " + originalData);
            
            for (PoolResult taskResultFromPool : taskResults) {
                List poolResult = (ArrayList) taskResultFromPool.getResultCollector().getResult();
                if (!partitioned) {
                    for (Object resultFromMember : poolResult) {
                        Map result = (HashMap) resultFromMember;
                        if (regionInfo.get("id").equals(result.get("memberId")))                   
                            continue;
                      
                        String memberId = (String) result.get("memberId");
                        Map memberData = (HashMap) result.get("map");
                        Map memberDataCopy = new HashMap();
                        memberDataCopy.putAll(memberData);
                        memberDataCopy.keySet().retainAll(originalData.keySet());
                        Map<String, Set> aggregationInfo = new HashMap<String, Set>();      
                        
                        Set keysForDiffValues = new HashSet();
                        Set absentKeys = new HashSet();
                        for (Object regionKey : memberDataCopy.keySet())
                            if (!memberDataCopy.get(regionKey).equals(originalData.get(regionKey)))      
                                keysForDiffValues.add(regionKey);

                        aggregationInfo.put("diffValues", keysForDiffValues);
                        
                        memberData.keySet().removeAll(memberDataCopy.keySet());

                        absentKeys.addAll(memberData.keySet());
                        absentKeys.addAll((HashSet) result.get("absentKeys"));
                        aggregationInfo.put("absentKeys", absentKeys);

                       
                         if (!clusterDifference.containsKey(memberId)) {
                            clusterDifference.put(memberId, aggregationInfo);
                        } else {
                            Map<String, Set> difference = clusterDifference.get(memberId);
                            difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys"));
                            difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                            clusterDifference.put(memberId, difference);
                        }
                    }
                } else {
                    Map targetMap = new HashMap();
                    Set absentPoolKeys = new HashSet();     
                    for (Object memberResult : poolResult) {
                        targetMap.putAll((Map) ((HashMap) memberResult).get("map"));
                        absentPoolKeys.addAll((Set) (((HashMap)memberResult).get("absentKeys")));
                    }
                    Map memberDataCopy = new HashMap();
                    memberDataCopy.putAll(targetMap);
                    memberDataCopy.keySet().retainAll(originalData.keySet());
                    Map<String, Set> aggregationInfo = new HashMap<String, Set>();      
                    
                    Set keysForDiffValues = new HashSet();
                    Set absentKeys = new HashSet();
                    for (Object regionKey : memberDataCopy.keySet())
                        if (!memberDataCopy.get(regionKey).equals(originalData.get(regionKey)))      
                            keysForDiffValues.add(regionKey);

                    aggregationInfo.put("diffValues", keysForDiffValues);
                    
                    targetMap.keySet().removeAll(memberDataCopy.keySet());

                    absentKeys.addAll(targetMap.keySet());
                    absentKeys.addAll(absentPoolKeys);
                    

                    if (!clusterDifference.containsKey(taskResultFromPool.getPool().toString())) {
                        clusterDifference.put(taskResultFromPool.getPool().toString(), aggregationInfo);
                    } else {
                        Map<String, Set> difference = clusterDifference.get(taskResultFromPool.getPool().toString());
                      
                        difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                        clusterDifference.put(taskResultFromPool.getPool().toString(), difference);
                    }
                }
            }

            taskResults.clear();
        }

        System.out.println("____________________________");
        System.out.println("difference: ");
        System.out.println(clusterDifference);
        executorService.shutdown();
        adminDs.disconnect();
    }
public static void main ( String [ ] args ) throws AdminException { AdminDistributedSystem adminDs = AdminDistributedSystemFactory . getDistributedSystem ( AdminDistributedSystemFactory . defineDistributedSystem ( ) ) ; adminDs . connect ( ) ; //connect to different clusters. test data                             todo: insert cli usage Pool poolForCluster1 = PoolManager . createFactory ( ) . addLocator ( "localhost" , 10330 ) . create ( "cluster1" ) ; Pool poolForCluster2 = PoolManager . createFactory ( ) . addLocator ( "localhost" , 10332 ) . create ( "cluster2" ) ; //Pool poolForServer =  PoolManager.createFactory().addServer("localhost", 40405).create("node"); List < Pool > poolList = new ArrayList < Pool > ( ) ; poolList . add ( poolForCluster1 ) ; poolList . add ( poolForCluster2 ) ; //poolList.add(poolForServer); //todo: insert checking that each cluster contains region and one's type is equal (Partitioned, Replicated) boolean partitioned = true ; //todo: insert CLI usage  + throw exception if real region has another type List < ServerLocation > serverFromPool = new ArrayList < ServerLocation > ( ) ; List < Pool > emptyPools = new ArrayList < Pool > ( ) ; //contains pool with no available servers for ( Pool pool : poolList ) { List < ServerLocation > allServers = null ; if ( ! pool . getLocators ( ) . isEmpty ( ) ) allServers = ( ( AutoConnectionSourceImpl ) ( ( PoolImpl ) pool ) . getConnectionSource ( ) ) . findAllServers ( ) ; else if ( ! pool . getServers ( ) . isEmpty ( ) ) allServers = Arrays . asList ( ( ( ( PoolImpl ) pool ) . getConnectionSource ( ) ) . findServer ( Collections . emptySet ( ) ) ) ; if ( allServers != null ) serverFromPool . addAll ( allServers ) ; else { logger . info ( "not found servers on locator {}" , pool ) ; emptyPools . add ( pool ) ; } } poolList . removeAll ( emptyPools ) ; if ( serverFromPool . size ( ) == 0 ) { logger . info ( "no servers available" ) ; return ; } for ( ServerLocation server : serverFromPool ) { System . out . println ( "host: " + server . getHostName ( ) ) ; System . out . println ( "port: " + server . getPort ( ) ) ; System . out . println ( "----------------------" ) ; } //source for comparison //todo: if this node doesn't contain region! it's problem Pool sourcePool ; if ( ! partitioned ) { int randomServerLocation = new Random ( ) . nextInt ( serverFromPool . size ( ) ) ; sourcePool = PoolManager . createFactory ( ) . addServer ( serverFromPool . get ( randomServerLocation ) . getHostName ( ) , serverFromPool . get ( randomServerLocation ) . getPort ( ) ) . create ( "target" ) ; } else { sourcePool = poolList . get ( 0 ) ; poolList . remove ( 0 ) ; } FunctionService . registerFunction ( new RegionInfoFunction ( ) ) ; ResultCollector regionInfoResult = FunctionService . onServers ( sourcePool ) . withArgs ( REGION_NAME ) . execute ( new RegionInfoFunction ( ) ) ; Map regionInfo = ( HashMap ) ( ( ArrayList ) regionInfoResult . getResult ( ) ) . get ( 0 ) ; System . out . println ( "region info: " + regionInfo ) ; int totalNumBuckets = ( Integer ) regionInfo . get ( "totalNumBuckets" ) ; //logger.debug("total keys' batch counts is ", totalNumBuckets); System . out . println ( "total keys' batch counts is " + totalNumBuckets ) ; KeyExtractor keyExtractor = new KeyExtractor ( REGION_NAME , sourcePool , partitioned , totalNumBuckets ) ; Map < String , Map < String , Set > > clusterDifference = new HashMap < String , Map < String , Set > > ( ) ; //key: memeberId list: absent keys, diff values //List<ResultCollector> taskResults = new ArrayList<ResultCollector>(); List < PoolResult > taskResults = new ArrayList < PoolResult > ( ) ; //List<Future<ResultCollector>> collectTasks = new ArrayList<Future<ResultCollector>>(poolList.size()); List < Future < PoolResult > > collectTasks = new ArrayList < Future < PoolResult > > ( poolList . size ( ) ) ; ExecutorService executorService = Executors . newFixedThreadPool ( poolList . size ( ) ) ; while ( keyExtractor . hasKeys ( ) ) { Set keys = keyExtractor . getNextKeysBatch ( ) ; System . out . println ( "keys to check: " + keys ) ; for ( Pool nextPool : poolList ) collectTasks . ( executorService . submit ( new CollectorTask ( keys , nextPool , REGION_NAME ) ) ) ; System . out . println ( "active tasks: " + collectTasks . size ( ) ) ; try { //for (Future<ResultCollector> futureTask : collectTasks) { for ( Future < PoolResult > futureTask : collectTasks ) { taskResults . add ( futureTask . get ( ) ) ; } } catch ( InterruptedException ie ) { ie . printStackTrace ( ) ; } catch ( ExecutionException ee ) { ee . printStackTrace ( ) ; } collectTasks . clear ( ) ; System . out . println ( "compare contents.." ) ; //getting original contents Map originalData = new HashMap ( ) ; /*           for (ResultCollector taskResultFromPool : taskResults) {
                List memberResult = (ArrayList) taskResultFromPool.getResult();
                if (!partitioned)
                    for (Object result : memberResult) {                        //if cluster has several nodes with replicated region... todo: adopt for partitioned
                        if (regionInfo.get("id").equals(((HashMap) result).get("memberId"))) {
                            originalData = (HashMap) ((HashMap) result).get("map");
                            break;
                        }
                    }
                else {
                    FutureTask<ResultCollector> ft = new FutureTask<ResultCollector>(new CollectorTask(keys, sourcePool, REGION_NAME));
                    try {
                        ResultCollector rc = ft.get();
                    } catch (Exception e) {
                        throw new RuntimeException("error getting key-hashcode from pool: " + sourcePool, e);
                    }
                }
            }*/ //FutureTask<ResultCollector> ft = new FutureTask<ResultCollector>(new CollectorTask(keys, sourcePool, REGION_NAME)); FutureTask < PoolResult > ft = new FutureTask < PoolResult > ( new CollectorTask ( keys , sourcePool , REGION_NAME ) ) ; ft . run ( ) ; try { //ResultCollector rc = ft.get(); PoolResult rc = ft . get ( ) ; List poolResult = ( List ) rc . getResultCollector ( ) . getResult ( ) ; for ( Object singleResult : poolResult ) { originalData . putAll ( ( Map ) ( ( HashMap ) singleResult ) . get ( "map" ) ) ; } } catch ( Exception e ) { throw new RuntimeException ( "error getting key-hash from pool: " + sourcePool , e ) ; } /*    if (true)
            return;

        //todo: aggregate members' data from one cluster
        if (partitioned)
            aggregateSingleClusterMemberData(taskResults);*/ System . out . println ( "original data is: " + originalData ) ; //for (ResultCollector taskResultFromPool : taskResults) { for ( PoolResult taskResultFromPool : taskResults ) { List poolResult = ( ArrayList ) taskResultFromPool . getResultCollector ( ) . getResult ( ) ; if ( ! partitioned ) { for ( Object resultFromMember : poolResult ) { Map result = ( HashMap ) resultFromMember ; if ( regionInfo . get ( "id" ) . equals ( result . get ( "memberId" ) ) ) //for replicated region continue ; /*  System.out.println("member id: " + result.get("memberId"));
                        System.out.println("absent keys is " + result.get("absentKeys"));
                        System.out.println("it's data is: " + result.get("map"));*/ String memberId = ( String ) result . get ( "memberId" ) ; Map memberData = ( HashMap ) result . get ( "map" ) ; Map memberDataCopy = new HashMap ( ) ; memberDataCopy . putAll ( memberData ) ; memberDataCopy . keySet ( ) . retainAll ( originalData . keySet ( ) ) ; Map < String , Set > aggregationInfo = new HashMap < String , Set > ( ) ; //contains aggregation info for a member //keys witch values are different Set keysForDiffValues = new HashSet ( ) ; Set absentKeys = new HashSet ( ) ; for ( Object regionKey : memberDataCopy . keySet ( ) ) if ( ! memberDataCopy . get ( regionKey ) . equals ( originalData . get ( regionKey ) ) ) //we compare not original values, but it's hash code keysForDiffValues . add ( regionKey ) ; aggregationInfo . put ( "diffValues" , keysForDiffValues ) ; //absent keys memberData . keySet ( ) . removeAll ( memberDataCopy . keySet ( ) ) ; absentKeys . addAll ( memberData . keySet ( ) ) ; absentKeys . addAll ( ( HashSet ) result . get ( "absentKeys" ) ) ; aggregationInfo . put ( "absentKeys" , absentKeys ) ; /* if (!clusterDifference.containsKey(memberId)) {
                            clusterDifference.put(memberId, aggregationInfo);
                        } else {
                            Map<String, Set> difference = clusterDifference.get(memberId);
                            difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys"));
                            difference.get("diffValues").addAll(aggregationInfo.get("diffValues"));
                            clusterDifference.put(memberId, difference);
                        }*/ if ( ! clusterDifference . containsKey ( memberId ) ) { clusterDifference . put ( memberId , aggregationInfo ) ; } else { Map < String , Set > difference = clusterDifference . get ( memberId ) ; difference . get ( "absentKeys" ) . addAll ( aggregationInfo . get ( "absentKeys" ) ) ; difference . get ( "diffValues" ) . addAll ( aggregationInfo . get ( "diffValues" ) ) ; clusterDifference . put ( memberId , difference ) ; } } } else { Map targetMap = new HashMap ( ) ; Set absentPoolKeys = new HashSet ( ) ; //aggregate absent keys from pool members for ( Object memberResult : poolResult ) { targetMap . putAll ( ( Map ) ( ( HashMap ) memberResult ) . get ( "map" ) ) ; absentPoolKeys . addAll ( ( Set ) ( ( ( HashMap ) memberResult ) . get ( "absentKeys" ) ) ) ; } Map memberDataCopy = new HashMap ( ) ; memberDataCopy . putAll ( targetMap ) ; memberDataCopy . keySet ( ) . retainAll ( originalData . keySet ( ) ) ; Map < String , Set > aggregationInfo = new HashMap < String , Set > ( ) ; //contains aggregation info for a member //keys witch values are different Set keysForDiffValues = new HashSet ( ) ; Set absentKeys = new HashSet ( ) ; for ( Object regionKey : memberDataCopy . keySet ( ) ) if ( ! memberDataCopy . get ( regionKey ) . equals ( originalData . get ( regionKey ) ) ) //we compare not original values, but it's hash code keysForDiffValues . add ( regionKey ) ; aggregationInfo . put ( "diffValues" , keysForDiffValues ) ; //absent keys targetMap . keySet ( ) . removeAll ( memberDataCopy . keySet ( ) ) ; absentKeys . addAll ( targetMap . keySet ( ) ) ; absentKeys . addAll ( absentPoolKeys ) ; //aggregationInfo.put("absentKeys", absentKeys); if ( ! clusterDifference . containsKey ( taskResultFromPool . getPool ( ) . toString ( ) ) ) { clusterDifference . put ( taskResultFromPool . getPool ( ) . toString ( ) , aggregationInfo ) ; } else { Map < String , Set > difference = clusterDifference . get ( taskResultFromPool . getPool ( ) . toString ( ) ) ; //  difference.get("absentKeys").addAll(aggregationInfo.get("absentKeys")); difference . get ( "diffValues" ) . addAll ( aggregationInfo . get ( "diffValues" ) ) ; clusterDifference . put ( taskResultFromPool . getPool ( ) . toString ( ) , difference ) ; } } } taskResults . clear ( ) ; } System . out . println ( "____________________________" ) ; System . out . println ( "difference: " ) ; System . out . println ( clusterDifference ) ; executorService . shutdown ( ) ; adminDs . disconnect ( ) ; }
todo: insert from cli
todo : insert from cli