code: public static Map<Integer, Integer> getMapOfContiguousPartitions(final Cluster cluster,
                                                                       int zoneId) {
          List<Integer> partitionIds = new ArrayList<Integer>(cluster.getPartitionIdsInZone(zoneId));
          Map<Integer, Integer> partitionIdToRunLength = Maps.newHashMap();
  
          if(partitionIds.isEmpty()) {
              return partitionIdToRunLength;
          }
  
          int lastPartitionId = partitionIds.get(0);
          int initPartitionId = lastPartitionId;
  
          for(int offset = 1; offset < partitionIds.size(); offset++) {
              int partitionId = partitionIds.get(offset);
              if(partitionId == lastPartitionId + 1) {
                  lastPartitionId = partitionId;
                  continue;
              }
              int runLength = lastPartitionId - initPartitionId + 1;
  
              partitionIdToRunLength.put(initPartitionId, runLength);
  
              initPartitionId = partitionId;
              lastPartitionId = initPartitionId;
          }
  
          int runLength = lastPartitionId - initPartitionId + 1;
          if(lastPartitionId == cluster.getNumberOfPartitions() - 1
             && partitionIdToRunLength.containsKey(0)) {
              // special case of contiguity that wraps around the ring.
              partitionIdToRunLength.put(initPartitionId, runLength + partitionIdToRunLength.get(0));
              partitionIdToRunLength.remove(0);
          } else {
              partitionIdToRunLength.put(initPartitionId, runLength);
          }
  
          return partitionIdToRunLength;
      }
docs: Determines run length for each 'initial' partition ID. Note that a
  contiguous run may "wrap around" the end of the ring.
  
  @param cluster
  @param zoneId
  @return map of initial partition Id to length of contiguous run of
  partition IDs within the same zone..