Decorates the given cache topology to add a key partitioner.

The key partitioner may include support for grouping as well.
491882
private CacheTopology addPartitioner(CacheTopology cacheTopology) {
      ConsistentHash currentCH = cacheTopology.getCurrentCH();
      currentCH = new PartitionerConsistentHash(currentCH, keyPartitioner);
      ConsistentHash pendingCH = cacheTopology.getPendingCH();
      if (pendingCH != null) {
         pendingCH = new PartitionerConsistentHash(pendingCH, keyPartitioner);
      }
      ConsistentHash unionCH = cacheTopology.getUnionCH();
      if (unionCH != null) {
         unionCH = new PartitionerConsistentHash(unionCH, keyPartitioner);
      }
      return new CacheTopology(cacheTopology.getTopologyId(), cacheTopology.getRebalanceId(), currentCH, pendingCH,
            unionCH, cacheTopology.getPhase(), cacheTopology.getActualMembers(), cacheTopology.getMembersPersistentUUIDs());
   }
262613
private void addSuccessor(final int stackSize, final Label successor) {
        Edge b;
        // creates a new Edge object or reuses one from the shared pool
        synchronized (SIZE) {
            if (pool == null) {
                b = new Edge();
            } else {
                b = pool;
                // removes b from the pool
                pool = pool.poolNext;
            }
        }
        // adds the previous Edge to the list of Edges used by this MethodWriter
        if (tail == null) {
            tail = b;
        }
        b.poolNext = head;
        head = b;
        // initializes the previous Edge object...
        b.stackSize = stackSize;
        b.successor = successor;
        // ...and adds it to the successor list of the currentBlock block
        b.next = currentBlock.successors;
        currentBlock.successors = b;
    }
1298842
public void addCacheConfiguration(String partition, Configuration<String, byte[]> configuration) {
		cacheConfigurations.put(partition, configuration);
	}
743506
def add_partition(self, partition):
        """"""
        assert(partition not in self._partitions)
        # Add partition to existing set
        self._partitions.add(partition)
        # Add broker to replica list
        partition.add_replica(self)
388460
public static <K, V> KeyAffinityService<K> newKeyAffinityService(Cache<K, V> cache, Executor ex, KeyGenerator<K> keyGenerator, int keyBufferSize) {
      return newKeyAffinityService(cache, ex, keyGenerator, keyBufferSize, true);
   }
929119
public StateSnapshot.StateKeyGroupWriter partitionByKeyGroup() {
		if (computedResult == null) {
			reportAllElementKeyGroups();
			int outputNumberOfElements = buildHistogramByAccumulatingCounts();
			executePartitioning(outputNumberOfElements);
		}
		return computedResult;
	}
670088
private static PersistenceStrategy<?, ?, ?> getPerKindStrategy(EmbeddedCacheManager externalCacheManager, URL configUrl, JtaPlatform platform) {
		OnePerKindKeyProvider keyProvider = new OnePerKindKeyProvider();

		OnePerKindCacheManager cacheManager = externalCacheManager != null ?
				new OnePerKindCacheManager( externalCacheManager ) :
				new OnePerKindCacheManager( configUrl, platform, keyProvider );

		return new PersistenceStrategy<EntityKey, AssociationKey, IdSourceKey>( cacheManager, keyProvider );
	}
804779
def partition_key(self, value):
        
        annotations = dict(self._annotations)
        annotations[self._partition_key] = value
        header = MessageHeader()
        header.durable = True
        self.message.annotations = annotations
        self.message.header = header
        self._annotations = annotations
155005
private void addConstructorByKey() {
    context.constructorByKey = MethodSpec.constructorBuilder().addParameter(keySpec);
    context.constructorByKey.addParameter(keyRefQueueSpec);
    addCommonParameters(context.constructorByKey);
    if (isBaseClass()) {
      callSiblingConstructor();
    } else {
      callParentByKey();
    }
  }
760349
public static Object groupKey(Object key, int groupLevel) {
        if (groupLevel > 0 && (key instanceof List) && (((List<Object>) key).size() > groupLevel)) {
            return ((List<Object>) key).subList(0, groupLevel);
        } else {
            return key;
        }
    }
974533
public V put(K key, V value) {
    return _cacheImpl.put(key, value);
  }
1068412
private void addCommonCacheKeyParts(StringBuilder builder) {
		builder.append("kind=").append(getKind());
		List<FilterPredicate> predicates = query.getFilterPredicates();
		if (predicates.size() > 0) {
			builder.append(",pred=").append(predicates);
		}
	}
167365
public <U extends UnsortedGrouping<?>> void add(int id, U set) {
		cacheSetType(id, SetType.UNSORTED_GROUPING);
		unsortedGroupings.put(id, set);
	}
703703
public CacheTopology computePreferredTopology(Map<Address, CacheStatusResponse> statusResponseMap) {
      List<Partition> partitions = computePartitions(statusResponseMap, "");

      return partitions.size() != 0 ? selectPreferredPartition(partitions).topology : null;
   }
378773
public void commit(String groupId, KafkaTopicPartitionOffset... tpoList) {
        KafkaMsgConsumer kafkaConsumer = cacheConsumers.get(groupId);
        if (kafkaConsumer != null) {
            kafkaConsumer.commit(tpoList);
        }
    }
1221511
@SuppressWarnings("unchecked")
  public final Cache2kBuilder<K, V> wrappingLoader(AdvancedCacheLoader<K, LoadDetail<V>> l) {
    config().setAdvancedLoader((
      CustomizationSupplier<AdvancedCacheLoader<K, V>>) (Object) wrapCustomizationInstance(l));
    return this;
  }
650345
private static <T> Iterable<List<T>> partitionExponentially(List<T> values, int minBatchSize, int maxBatchSize)
    {
        return () -> new AbstractIterator<List<T>>()
        {
            private int currentSize = minBatchSize;
            private final Iterator<T> iterator = values.iterator();

            @Override
            protected List<T> computeNext()
            {
                if (!iterator.hasNext()) {
                    return endOfData();
                }

                int count = 0;
                ImmutableList.Builder<T> builder = ImmutableList.builder();
                while (iterator.hasNext() && count < currentSize) {
                    builder.add(iterator.next());
                    ++count;
                }

                currentSize = min(maxBatchSize, currentSize * 2);
                return builder.build();
            }
        };
    }