  public Collection<Partition<RedisOutputOperator<K,V>>> definePartitions(Collection<Partition<RedisOutputOperator<K,V>>> partitions, int incrementalCapacity)
  {
    Collection c = partitions;
    Collection<Partition<RedisOutputOperator<K, V>>> operatorPartitions = c;
    Partition<RedisOutputOperator<K, V>> template = null;
    Iterator<Partition<RedisOutputOperator<K, V>>> itr = operatorPartitions.iterator();
    template = itr.next();
    String[] connectionArr = connectionList.trim().split("\\|");
    int size = connectionArr.length;
    if (size > (incrementalCapacity + operatorPartitions.size()))
      size = incrementalCapacity + operatorPartitions.size();
    int partitionBits = (Integer.numberOfLeadingZeros(0) - Integer.numberOfLeadingZeros(size - 1));
    int partitionMask = 0;
    if (partitionBits > 0) {
      partitionMask = -1 >>> (Integer.numberOfLeadingZeros(-1)) - partitionBits;
    }
    LOG.debug("partition mask {}", partitionMask);
    ArrayList<Partition<RedisOutputOperator<K,V>>> operList = new ArrayList<Partitionable.Partition<RedisOutputOperator<K,V>>>(size);
    while (size > 0) {
      size--;
      RedisOutputOperator<K, V> opr = new RedisOutputOperator<K, V>();
      opr.setConnectionList(connectionArr[size].trim());
      opr.setKeyExpiryTime(keyExpiryTime);
      opr.setTimeout(timeout);
      opr.setContinueOnError(continueOnError);
      opr.setName(getName());
      Partition<RedisOutputOperator<K, V>> p = new DefaultPartition(opr);
      operList.add(p);
    }
    for (int i = 0; i <= partitionMask; i++) {
      Partition<RedisOutputOperator<K, V>> p = operList.get(i % operList.size());
      PartitionKeys pks = p.getPartitionKeys().get(input);
      if (pks == null) {
        p.getPartitionKeys().put(input, new PartitionKeys(partitionMask, Sets.newHashSet(i)));
      } else {
        pks.partitions.add(i);
      }
      pks = p.getPartitionKeys().get(inputInd);
      if (pks == null) {
        p.getPartitionKeys().put(inputInd, new PartitionKeys(partitionMask, Sets.newHashSet(i)));
      } else {
        pks.partitions.add(i);
      }
    }
    return operList;
  }

  public Collection<Partition<RedisOutputOperator<K,V>>> definePartitions(Collection<Partition<RedisOutputOperator<K,V>>> partitions, int incrementalCapacity)
  {
    Collection c = partitions;
    Collection<Partition<RedisOutputOperator<K, V>>> operatorPartitions = c;
    Partition<RedisOutputOperator<K, V>> template = null;
    Iterator<Partition<RedisOutputOperator<K, V>>> itr = operatorPartitions.iterator();
    template = itr.next();
    if(connectionList == null){
      connectionList = host+":"+port+","+dbIndex;
    }
    String[] connectionArr = connectionList.trim().split("\\|");
    int size = connectionArr.length;
    if (size > (incrementalCapacity + operatorPartitions.size()))
      size = incrementalCapacity + operatorPartitions.size();
    int partitionBits = (Integer.numberOfLeadingZeros(0) - Integer.numberOfLeadingZeros(size - 1));
    int partitionMask = 0;
    if (partitionBits > 0) {
      partitionMask = -1 >>> (Integer.numberOfLeadingZeros(-1)) - partitionBits;
    }
    LOG.debug("partition mask {}", partitionMask);
    ArrayList<Partition<RedisOutputOperator<K,V>>> operList = new ArrayList<Partitionable.Partition<RedisOutputOperator<K,V>>>(size);
    while (size > 0) {
      size--;
      RedisOutputOperator<K, V> opr = new RedisOutputOperator<K, V>();
      opr.setConnectionList(connectionArr[size].trim());
      opr.setKeyExpiryTime(keyExpiryTime);
      opr.setTimeout(timeout);
      opr.setContinueOnError(continueOnError);
      opr.setName(getName());
      Partition<RedisOutputOperator<K, V>> p = new DefaultPartition(opr);
      operList.add(p);
    }
    for (int i = 0; i <= partitionMask; i++) {
      Partition<RedisOutputOperator<K, V>> p = operList.get(i % operList.size());
      PartitionKeys pks = p.getPartitionKeys().get(input);
      if (pks == null) {
        p.getPartitionKeys().put(input, new PartitionKeys(partitionMask, Sets.newHashSet(i)));
      } else {
        pks.partitions.add(i);
      }
      pks = p.getPartitionKeys().get(inputInd);
      if (pks == null) {
        p.getPartitionKeys().put(inputInd, new PartitionKeys(partitionMask, Sets.newHashSet(i)));
      } else {
        pks.partitions.add(i);
      }
    }
    return operList;
  }
