icegem-cache-utils/src/main/java/com/googlecode/icegem/cacheutils/regioncomparator/HashCodeCollector.java
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();

        //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();
    }
todo: insert from cli