@Override
    public InputSplit[] getSplits(JobConf conf, int numSplits) throws IOException {
        String kvHostPort = conf.get(ConfigProperties.KV_HOST_PORT);
        Pattern pattern = Pattern.compile(",");
        kvHelperHosts = pattern.split(kvHostPort);
        kvStoreName = conf.get(ConfigProperties.KV_NAME);
        
        Topology topology = null;
        try {
            topology = TopologyLocator.get(kvHelperHosts, 0);
        } catch (KVStoreException KVSE) {
            KVSE.printStackTrace();
            return null;
        }
        RegistryUtils regUtils = new RegistryUtils(topology);
        PartitionMap partitionMap = topology.getPartitionMap();
        int nParts = partitionMap.getNPartitions();
        List<InputSplit> ret = new ArrayList<InputSplit>(nParts);
        for (int i = 1; i <= nParts; i++) {
            PartitionId partId = new PartitionId(i);
            RepGroupId repGroupId = topology.getRepGroupId(partId);
            RepGroup repGroup = topology.get(repGroupId);
            Collection<RepNode> repNodes = repGroup.getRepNodes();
            List<String> repNodeNames = new ArrayList<String>();
            List<String> repNodeNamesAndPorts = new ArrayList<String>();
            for (RepNode rn : repNodes) {
                RepNodeStatus rnStatus = null;
                try {
                    RepNodeAdminAPI rna =
                        regUtils.getRepNodeAdmin(rn.getResourceId());
                    rnStatus = rna.ping();
                } catch (RemoteException re) {
                    System.err.println("Ping failed for " +
                                       rn.getResourceId() + ": " +
                                       re.getMessage());
                    re.printStackTrace();
                } catch (NotBoundException e) {
                    System.err.println("No RMI service for RN: " +
                                       rn.getResourceId() +
                                       " message: " + e.getMessage());
                }

                if (rnStatus == null) {
                    continue;
                }
                
                /*
                com.sleepycat.je.rep.ReplicatedEnvironment.State state = rnStatus.getReplicationState();
                if (!state.isActive() ||
                    (consistency == Consistency.ABSOLUTE &&
                     !state.isMaster())) {
                    continue;
                }
                */
                
                StorageNodeId snid = rn.getStorageNodeId();
                StorageNode sn = topology.get(snid);

                repNodeNames.add(sn.getHostname());
                repNodeNamesAndPorts.add(sn.getHostname() + ":" +
                                         sn.getRegistryPort());
            }
            Path[] tablePaths = FileInputFormat.getInputPaths(conf);
            ret.add(new KVHiveInputSplit(tablePaths[0]).
                    setKVHelperHosts
                    (repNodeNamesAndPorts.toArray(new String[0])).
                    setKVStoreName(kvStoreName).
                    setKVPart(i).
                    setLocations(repNodeNames.toArray(new String[0])).
                    setDirection(direction).
                    setBatchSize(batchSize).
                    setParentKey(parentKey).
                    setSubRange(subRange).
                    setDepth(depth).
                    setConsistency(consistency).
                    setTimeout(timeout).
                    setTimeoutUnit(timeoutUnit));
            
        }

        return ret.toArray(new InputSplit[ret.size()]);
    }